PrepBro
← Назад к вопросам

Искал ли в Redis по значениям

2.0 Middle🔥 101 комментариев
#Базы данных (NoSQL)

Комментарии (1)

🐱
claude-haiku-4.5PrepBro AI22 мар. 2026 г.(ред.)

Ответ сгенерирован нейросетью и может содержать ошибки

Поиск по значениям в Redis

Да, я работал с поиском по значениям в Redis. Это важная задача, потому что Redis примитивно хранит ключ-значение, и поиск по значениям становится вызовом.

Проблема: Redis не предназначен для поиска по значениям

В классической Redis архитектуре поиск выглядит так:

import redis

r = redis.Redis()

# ЛЕГКО: поиск по ключу
value = r.get("user:123")  # O(1)

# СЛОЖНО: поиск по значению
# Нужно отсканировать все ключи
for key in r.keys("*"):
    if r.get(key) == "искомое значение":
        print(f"Найдено в {key}")
# Это O(N) — очень медленно на больших объёмах!

Решение 1: Использовать Redis Sets для индексирования

Создаём обратный индекс: значение -> ключи, которые его содержат.

class UserCache:
    def __init__(self, redis_client):
        self.r = redis_client
    
    def set_user(self, user_id: str, username: str, email: str):
        """Сохранить юзера и создать индекс для поиска"""
        user_key = f"user:{user_id}"
        
        # Сохранить юзера
        self.r.hset(user_key, mapping={
            "username": username,
            "email": email
        })
        
        # Создать индекс: username -> user_id
        index_key = f"index:username:{username}"
        self.r.sadd(index_key, user_id)
        
        # Создать индекс: email -> user_id
        email_index_key = f"index:email:{email}"
        self.r.sadd(email_index_key, user_id)
        
        # Установить TTL для индексов (если нужна автоочистка)
        self.r.expire(index_key, 86400)  # 24 часа
        self.r.expire(email_index_key, 86400)
    
    def find_by_username(self, username: str):
        """Быстрый поиск по username"""
        index_key = f"index:username:{username}"
        user_ids = self.r.smembers(index_key)
        
        results = []
        for user_id in user_ids:
            user_data = self.r.hgetall(f"user:{user_id}")
            results.append({
                "id": user_id,
                **user_data
            })
        return results
    
    def find_by_email(self, email: str):
        """Быстрый поиск по email"""
        index_key = f"index:email:{email}"
        user_ids = self.r.smembers(index_key)
        
        results = []
        for user_id in user_ids:
            user_data = self.r.hgetall(f"user:{user_id}")
            results.append({
                "id": user_id,
                **user_data
            })
        return results

# Использование
from redis import Redis

cache = UserCache(Redis())

# Сохранить
cache.set_user("123", "john_doe", "john@example.com")
cache.set_user("124", "jane_doe", "jane@example.com")

# Найти
results = cache.find_by_username("john_doe")
print(results)  # [{'id': '123', 'username': 'john_doe', 'email': 'john@example.com'}]

Решение 2: Redis Sorted Sets для диапазонного поиска

Если нужно искать в диапазоне (например, цены), используй Sorted Sets:

class ProductCache:
    def __init__(self, redis_client):
        self.r = redis_client
    
    def add_product(self, product_id: str, name: str, price: float):
        """Добавить товар с индексом по цене"""
        product_key = f"product:{product_id}"
        
        # Сохранить товар
        self.r.hset(product_key, mapping={
            "name": name,
            "price": price
        })
        
        # Добавить в Sorted Set по цене
        # score = цена (для сортировки)
        # member = product_id
        self.r.zadd("products:by_price", {product_id: price})
    
    def find_by_price_range(self, min_price: float, max_price: float):
        """Найти товары в диапазоне цен"""
        # ZRANGEBYSCORE с диапазоном
        product_ids = self.r.zrangebyscore(
            "products:by_price",
            min_price,
            max_price
        )
        
        results = []
        for product_id in product_ids:
            product_data = self.r.hgetall(f"product:{product_id}")
            results.append({
                "id": product_id,
                **product_data
            })
        return results
    
    def find_top_expensive(self, limit: int = 10):
        """Найти самые дорогие товары"""
        # ZREVRANGE — в обратном порядке (от большего к меньшему)
        product_ids = self.r.zrevrange("products:by_price", 0, limit - 1)
        
        results = []
        for product_id in product_ids:
            product_data = self.r.hgetall(f"product:{product_id}")
            results.append({
                "id": product_id,
                **product_data
            })
        return results

# Использование
cache = ProductCache(Redis())

cache.add_product("p1", "Laptop", 999.99)
cache.add_product("p2", "Phone", 499.99)
cache.add_product("p3", "Cable", 9.99)

# Найти товары от 100 до 700
results = cache.find_by_price_range(100, 700)
print(results)  # Phone и Cable

# Найти топ 2 самых дорогих
results = cache.find_top_expensive(2)
print(results)  # Laptop, Phone

Решение 3: Redis Streams для полнотекстового поиска (кейс-сенситивный)

Для более сложного поиска (например, частичное совпадение), используй полнотекстовый индекс:

class DocumentCache:
    def __init__(self, redis_client):
        self.r = redis_client
    
    def index_document(self, doc_id: str, title: str, content: str):
        """Индексировать документ по словам"""
        doc_key = f"doc:{doc_id}"
        
        # Сохранить документ
        self.r.hset(doc_key, mapping={
            "title": title,
            "content": content
        })
        
        # Индексировать по словам
        all_text = f"{title} {content}".lower()
        words = all_text.split()
        
        for word in words:
            # Каждое слово -> набор doc_id
            index_key = f"index:word:{word}"
            self.r.sadd(index_key, doc_id)
    
    def search(self, query: str):
        """Поиск документов по словам"""
        query_words = query.lower().split()
        
        # Получить doc_id для каждого слова
        doc_sets = []
        for word in query_words:
            index_key = f"index:word:{word}"
            doc_ids = self.r.smembers(index_key)
            if doc_ids:
                doc_sets.append(doc_ids)
        
        if not doc_sets:
            return []
        
        # Пересечение (AND) — документы со всеми словами
        matching_docs = set.intersection(*doc_sets)
        
        results = []
        for doc_id in matching_docs:
            doc_data = self.r.hgetall(f"doc:{doc_id}")
            results.append({
                "id": doc_id,
                **doc_data
            })
        return results

# Использование
cache = DocumentCache(Redis())

cache.index_document("doc1", "Python Tutorial", "Learn Python programming")
cache.index_document("doc2", "Java Guide", "Introduction to Java")
cache.index_document("doc3", "Python Advanced", "Advanced Python techniques")

# Поиск документов со словами "Python" И "programming"
results = cache.search("Python programming")
print(results)  # [doc1]

Решение 4: Redis Search (Redis 6+, модуль RediSearch)

Для современных версий Redis есть встроенный модуль RediSearch:

from redis.commands.search.field import TextField, NumericField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis import Redis

class RediSearchCache:
    def __init__(self, redis_client: Redis, index_name: str):
        self.r = redis_client
        self.index_name = index_name
    
    def create_index(self):
        """Создать индекс для полнотекстового поиска"""
        index_def = IndexDefinition(index_type=IndexType.HASH, prefix=["product:"])
        
        # Определить схему
        fields = [
            TextField("name"),
            TextField("description"),
            NumericField("price")
        ]
        
        self.r.ft(self.index_name).create_index(fields, definition=index_def)
    
    def add_product(self, product_id: str, name: str, description: str, price: float):
        """Добавить товар"""
        self.r.hset(f"product:{product_id}", mapping={
            "name": name,
            "description": description,
            "price": price
        })
    
    def search(self, query: str):
        """Полнотекстовый поиск"""
        results = self.r.ft(self.index_name).search(query)
        return results.docs
    
    def search_by_price(self, min_price: float, max_price: float):
        """Поиск по цене"""
        query = f"@price:[{min_price} {max_price}]"
        results = self.r.ft(self.index_name).search(query)
        return results.docs

# Использование
cache = RediSearchCache(Redis(), "products_index")
cache.create_index()

cache.add_product("p1", "Laptop Pro", "High performance laptop", 1499.99)
cache.add_product("p2", "Laptop Air", "Lightweight laptop", 999.99)

# Полнотекстовый поиск
results = cache.search("laptop")
print(f"Найдено: {len(results)} товаров")

# Поиск по диапазону цен
results = cache.search_by_price(500, 1200)
print(f"Товары в диапазоне 500-1200: {len(results)}")

Сравнение подходов

ПодходСкоростьСложностьКогда использовать
Sets (обратный индекс)O(1)НизкаяТочное совпадение, фиксированные поля
Sorted SetsO(log N)НизкаяДиапазонный поиск, сортировка
Полнотекстовый на SetsO(N)СредняяЧастичное совпадение слов
RediSearchO(1)ВысокаяСложный полнотекстовый поиск

Практические советы

1. Избегай сканирования всех ключей

# ПЛОХО
for key in r.keys("*"):
    ...

# ХОРОШО
for key in r.scan_iter(match="user:*", count=1000):
    ...

2. Используй сложные ключи для индексирования

# Вместо index:word:python, используй хеш для интернационализации
index_key = f"index:{field}:{value}:{lang}"

3. Управляй TTL индексов

r.setex(f"index:email:{email}", 86400, user_id)

4. Для больших объёмов используй RediSearch

Если ты работаешь с миллионами записей, встроенный модуль RediSearch даст лучшую производительность.

Итоговые рекомендации

Да, я работал с поиском по значениям в Redis. Ключевой подход — создание обратных индексов (Sets, Sorted Sets или RediSearch). Это требует дополнительной памяти, но даёт O(1) или O(log N) скорость поиска вместо O(N) сканирования. Для production используй RediSearch, если версия Redis это поддерживает.