Introdução às Estruturas de Dados Fundamentais do Redis

A força do Redis reside em suas diversas estruturas de dados, projetadas para abordar uma ampla gama de cenários de negócios de forma eficiente. Este artigo foca nas duas mais básicas e frequentemente utilizadas: Strings e Hashes (Mapas de Hash).

Visão Geral das Estruturas

• String: É o tipo mais simples, onde uma chave corresponde a um único valor. Embora fundamentalmente um string, pode armazenar qualquer dado binário seguro, como textos, números, JSON serializado ou até mesmo imagens. É a base para muitas funcionalidades complexas.
• Hash: Representa um mapa de chave-valor onde uma chave Redis principal aponta para um conjunto de pares campo-valor. É ideal para armazenar objetos (por exemplo, perfis de usuário, atributos de produto), permitindo o acesso e a modificação de campos individuais sem a necessidade de manipular o objeto inteiro.

Compreender suas implementações subjacentes e casos de uso adequados é crucial para desenvolver aplicações Redis de alto desempenho.

Configuração do Ambiente Python

Para interagir com o Redis, utilizaremos a biblioteca redis-py. Certifique-se de que você já a tenha instalada e configurada para se conectar ao seu servidor Redis. O exemplo a seguir demonstra a configuração de um cliente Redis utilizando um pool de conexões, que é a prática recomendada:

import os
import redis
from redis import Redis

# Configurações do pool de conexões
pool_config = redis.ConnectionPool(
    host=os.getenv('REDIS_HOST', 'localhost'),
    port=int(os.getenv('REDIS_PORT', 6379)),
    password=os.getenv('REDIS_PASSWORD'),  # Comente esta linha se não houver senha
    decode_responses=True,  # Decodifica automaticamente bytes para strings
    max_connections=10
)
conexao_redis = Redis(connection_pool=pool_config)

# Teste de conexão simples
if conexao_redis.ping():
    print("Conexão com Redis estabelecida com sucesso!")
else:
    print("Falha ao conectar ao Redis.")

Operações com Strings

As Strings são extremamente versáteis e podem ser usadas para caching, contadores, bloqueios distribuídos e muito mais.

Operações Básicas com Strings

Este bloco de código ilustra as operações fundamentais para armazenar e recuperar dados de String.

def demonstrar_operacoes_string_basicas(cliente_redis: Redis):
    """Demonstra operações básicas com Strings: armazenar, recuperar, expirar e definir se não existe."""
    print("\n--- Operações Básicas de String ---")

    # 1. Definir e Obter um valor (SET/GET)
    # Cenário: cache simples de informações
    cliente_redis.set('nome_usuario', 'fernanda')
    nome_recuperado = cliente_redis.get('nome_usuario')
    print(f"Nome de Usuário: {nome_recuperado}")

    # 2. Definir com tempo de expiração (SETEX)
    # Cenário: códigos de verificação de celular, sessões temporárias
    cliente_redis.setex('cod_sms:987654321', 180, '987654') # Expira em 180 segundos
    codigo_sms = cliente_redis.get('cod_sms:987654321')
    print(f"Código SMS: {codigo_sms}")
    tempo_restante = cliente_redis.ttl('cod_sms:987654321') # Verifica o tempo de vida restante
    print(f"TTL do Código SMS: {tempo_restante} segundos")

    # 3. Definir apenas se a chave não existe (SETNX)
    # Cenário: implementação de bloqueios distribuídos ou inicialização única
    foi_definido = cliente_redis.setnx('app_inicializado', 'sim')
    if foi_definido:
        print("Marcador de inicialização do aplicativo definido com sucesso.")
    else:
        print("Aplicativo já foi inicializado anteriormente.")
    
    # 4. Operações em Massa (MSET/MGET) - Reduz o número de viagens de rede
    # Cenário: carregar/salvar múltiplas configurações ou estados de uma vez
    configs_app = {"config:tema": "claro", "config:idioma": "pt-BR", "config:notificacoes": "off"}
    cliente_redis.mset(configs_app)
    
    chaves_para_buscar = ["config:tema", "config:idioma", "config:notificacoes", "config:inexistente"]
    valores_configs = cliente_redis.mget(chaves_para_buscar)
    print(f"Configurações em lote: {valores_configs}") # Retorna uma lista com os valores correspondentes ou None

# Chame a função para demonstrar
demonstrar_operacoes_string_basicas(conexao_redis)

Operações Numéricas com Strings (Contadores)

Strings podem ser tratadas como números para operações atômicas de incremento e decremento.

def demonstrar_contadores_string(cliente_redis: Redis):
    """Demonstra operações de contador com Strings."""
    print("\n--- Operações Numéricas com Strings ---")

    # Inicializa um contador
    cliente_redis.set('visualizacoes_pagina', 0)

    # 1. Incrementar (INCR/INCRBY)
    # Cenário: contagem de visualizações de artigo, likes de usuário, estoque de produtos
    total_views = cliente_redis.incr('visualizacoes_pagina') # Incrementa em 1, retorna 1
    total_views = cliente_redis.incr('visualizacoes_pagina') # Incrementa em 1, retorna 2
    total_views = cliente_redis.incrby('visualizacoes_pagina', 5) # Incrementa em 5, retorna 7
    print(f"Total de Visualizações da Página: {total_views}")

    # 2. Decrementar (DECR/DECRBY)
    # Cenário: dedução de estoque, cancelamento de operações
    estoque_atual = cliente_redis.set('produto:XPT:estoque', 100)
    estoque_atual = cliente_redis.decrby('produto:XPT:estoque', 15) # Decrementa em 15
    print(f"Estoque do Produto XPT: {estoque_atual}")

    # 3. Operações com Números de Ponto Flutuante (INCRBYFLOAT)
    # Cenário: valores monetários, pontuações, pesos
    cliente_redis.set('saldo_conta:cliente1', 500.25)
    novo_saldo = cliente_redis.incrbyfloat('saldo_conta:cliente1', 75.50) # Adiciona 75.50
    print(f"Novo Saldo da Conta: {novo_saldo}") # 575.75

# Chame a função para demonstrar
demonstrar_contadores_string(conexao_redis)

Operações com Hashes (Mapas de Hash)

Hashes são ideais para representar objetos com múltiplos atributos, permitindo operações atômicas em campos individuais.

Operações Básicas com Hashes

def demonstrar_operacoes_hash_basicas(cliente_redis: Redis):
    """Demonstra operações básicas com Hashes: definir, obter campos, listar e excluir."""
    print("\n--- Operações Básicas de Hash ---")
    id_usuario = "u12345"
    chave_usuario = f'usuario:{id_usuario}'

    # 1. Definir e Obter campos individuais (HSET/HGET)
    # Cenário: armazenar atributos de um usuário
    cliente_redis.hset(chave_usuario, 'nome', 'Carlos Eduardo')
    cliente_redis.hset(chave_usuario, 'email', 'carlos.edu@exemplo.com')
    nome_usuario = cliente_redis.hget(chave_usuario, 'nome')
    print(f"Nome do Usuário: {nome_usuario}")

    # 2. Definir múltiplos campos (HSET com mapeamento)
    # Nota: HMSET é considerado obsoleto; HSET com um dicionário é a abordagem moderna.
    # Cenário: definir todos os atributos de um objeto de uma vez
    dados_adicionais_usuario = {
        'idade': '28', # Valores de campos Hash são sempre strings
        'cidade': 'São Paulo',
        'profissao': 'Desenvolvedor'
    }
    cliente_redis.hset(chave_usuario, mapping=dados_adicionais_usuario)

    # 3. Obter múltiplos campos (HMGET)
    # Cenário: recuperar um subconjunto de atributos de um objeto
    campos_para_buscar = ['nome', 'email', 'idade', 'pais'] # 'pais' não existe
    info_usuario_lista = cliente_redis.hmget(chave_usuario, campos_para_buscar)
    print(f"Informações do Usuário (lista): {info_usuario_lista}") # ['Carlos Eduardo', 'carlos.edu@exemplo.com', '28', None]

    # 4. Obter todos os campos e valores (HGETALL)
    # CUIDADO: Use com moderação em Hashes grandes, pode consumir muita largura de banda e bloquear o servidor.
    todos_dados_usuario = cliente_redis.hgetall(chave_usuario)
    print(f"Todos os dados do Usuário (dict): {todos_dados_usuario}")

    # 5. Obter todos os nomes de campos ou todos os valores (HKEYS/HVALS)
    nomes_campos = cliente_redis.hkeys(chave_usuario)
    valores_campos = cliente_redis.hvals(chave_usuario)
    print(f"Nomes dos campos: {nomes_campos}")
    print(f"Valores dos campos: {valores_campos}")

    # 6. Verificar existência de campo (HEXISTS) e Excluir campo (HDEL)
    if cliente_redis.hexists(chave_usuario, 'email'):
        print("Campo 'email' existe.")
    
    cliente_redis.hdel(chave_usuario, 'profissao') # Exclui um campo
    print(f"Campos após exclusão: {cliente_redis.hkeys(chave_usuario)}")

# Chame a função para demonstrar
demonstrar_operacoes_hash_basicas(conexao_redis)

Operações Numéricas com Campos de Hash

Assim como as Strings, os campos de um Hash podem ser incrementados e decrementados atomicamente.

def demonstrar_contadores_hash(cliente_redis: Redis):
    """Demonstra operações numéricas em campos de Hash."""
    print("\n--- Operações Numéricas em Campos de Hash ---")
    id_produto = "prod3001"
    chave_produto = f'produto:{id_produto}'

    # Inicialização de campos
    cliente_redis.hset(chave_produto, 'preco', '125.50')
    cliente_redis.hset(chave_produto, 'contagem_vendas', '0')

    # Incrementar/Decrementar campos de Hash (HINCRBY/HINCRBYFLOAT)
    # Cenário: ajustar preços de produtos, contadores independentes por produto
    novas_vendas = cliente_redis.hincrby(chave_produto, 'contagem_vendas', 10) # Incrementa o campo 'contagem_vendas' em 10
    novo_preco = cliente_redis.hincrbyfloat(chave_produto, 'preco', -20.25) # Decrementa o campo 'preco' em 20.25
    print(f"Vendas do Produto: {novas_vendas}, Novo Preço: {novo_preco}")

# Chame a função para demonstrar
demonstrar_contadores_hash(conexao_redis)

Escolha entre String e Hash para Armazenar Objetos

Uma decisão comum é como representar objetos complexos no Redis. Para objetos aninhados, a extensão RedisJSON é uma solução robusta (fora do escopo deste artigo). No entanto, para objetos mais planos, a escolha entre Strings (com serialização JSON) e Hashes é fundamental.

Utilizando String (com JSON)

import json

def armazenar_objeto_string_json(cliente_redis: Redis):
    """Armazena um objeto como String JSON."""
    dados_perfil = {'nome': 'Mariana', 'idade': 25, 'cidade': 'Rio de Janeiro'}
    chave_perfil = 'perfil:mariana'

    # Escrita: Serializa e armazena o objeto inteiro
    cliente_redis.set(chave_perfil, json.dumps(dados_perfil))
    print(f"\nObjeto armazenado como String JSON para '{chave_perfil}'")

    # Leitura: Recupera e desserializa o objeto inteiro
    dados_str = cliente_redis.get(chave_perfil)
    if dados_str:
        perfil_recuperado = json.loads(dados_str)
        print(f"Perfil recuperado (String JSON): {perfil_recuperado}")
        print(f"Idade: {perfil_recuperado['idade']}")

    # Desvantagem: Para atualizar a idade, todo o objeto precisa ser lido, modificado e escrito de volta.
    # Não há atualização atômica de campo individual.
    # Exemplo de atualização:
    if dados_str:
        perfil_modificado = json.loads(dados_str)
        perfil_modificado['idade'] = 26
        cliente_redis.set(chave_perfil, json.dumps(perfil_modificado))
        print("Idade atualizada via String JSON.")

# Chame a função para demonstrar
armazenar_objeto_string_json(conexao_redis)

• Vantagens: Simples, pode armazenar estruturas aninhadas complexas de forma nativa no JSON.
• Desvantagens: Qualquer modificação, mesmo de um único atributo, exige a leitura, desserialização, modificação e serialização/escrita de todo o objeto. Isso gera sobrecarga de CPU e rede, e as operações não são atômicas para campos individuais.

Utilizando Hash

def armazenar_objeto_hash(cliente_redis: Redis):
    """Armazena um objeto utilizando o tipo Hash."""
    chave_perfil = 'perfil:gabriel'
    dados_perfil = {'nome': 'Gabriel', 'idade': '30', 'cidade': 'Belo Horizonte'} # Valores são strings!

    # Escrita: Define múltiplos campos de uma vez
    cliente_redis.hset(chave_perfil, mapping=dados_perfil)
    print(f"\nObjeto armazenado como Hash para '{chave_perfil}'")

    # Leitura de um único campo (altamente eficiente)
    nome = cliente_redis.hget(chave_perfil, 'nome')
    print(f"Nome do perfil (Hash): {nome}")

    # Atualização de um único campo (atômica e eficiente)
    cliente_redis.hset(chave_perfil, 'idade', '31')
    print(f"Idade do perfil atualizada (Hash). Nova idade: {cliente_redis.hget(chave_perfil, 'idade')}")

    # Recuperar todos os dados (cuidado com Hashes grandes)
    todos_dados = cliente_redis.hgetall(chave_perfil)
    print(f"Todos os dados do perfil (Hash): {todos_dados}")

# Chame a função para demonstrar
armazenar_objeto_hash(conexao_redis)

• Vantagens: Permite acesso e modificação atômicos e independentes de campos individuais, o que é muito eficiente. Otimizado para memória em Hashes pequenos via codificação ziplist.
• Desvantagens: Não suporta estruturas aninhadas diretamente (todos os valores dos campos são strings). Requer conversão de tipo no cliente (ex: int(valor)).

Conclusão da Escolha: Para objetos "planos" (sem aninhamento complexo) que exigem atualizações frequentes e parciais, Hashes são a escolha superior. Para objetos com estrutura compleexa, onde a leitura/escrita completa é aceitável, Strings com JSON são uma opção. No entanto, sempre considere alternativas como RedisJSON para modelos de dados mais ricos.

Otimização de Memória: Codificação Ziplist

O Redis otimiza o uso de memória para Hashes (e também para Listas, Sets e Sorted Sets) pequenos utilizando uma representação mais compacta chamada ziplist (lista compactada). Esta codificação consome menos memória do que uma tabela de hash tradicional. Um Hash é codificado como ziplist enquanto o número de elementos e o tamanho dos valores dos campos não excederem certos limites configuráveis:

• hash-max-ziplist-entries: Número máximo de campos em um Hash (padrão: 512).
• hash-max-ziplist-value: Tamanho máximo de um valor de campo (padrão: 64 bytes).

Se esses limites forem excedidos, o Redis converterá automaticamente o Hash para a codificação de tabela de hash, que é mais flexível, mas consome mais memória. Ajustar esses parâmetros no redis.conf pode ajudar a equilibrar o uso de memória e desempenho.

Considerações de Performance e Confiabilidade

Operações em Lote para Redução de RTT

A latência da rede é um fator significativo no desempenho. Comandos como MSET/MGET para Strings e HSET (com mapping)/HMGET para Hashes permitem executar múltiplas operações com uma única viagem de rede (RTT - Round Trip Time), resultando em ganhos substanciais de performance.

Risco de "Big Keys" (Chaves Grandes)

Evite armazenar Strings gigantes (geralmente acima de 10KB) ou Hashes com milhares de campos. "Big Keys" podem causar lentidão durante operações como persistência de dados (RDB/AOF), replicação ou exclusão, potencialmente bloqueando o servidor Redis. Para Hashes, use HLEN para monitorar o número de campos.

Cuidado com Comandos de Complexidade O(n)

• Comandos como HGETALL, HKEYS e HVALS possuem complexidade O(n), onde 'n' é o número de campos no Hash. Em Hashes muito grandes, esses comandos podem ser lentos e impactar o desempenho do servidor. Prefira HGET ou HMGET para buscar apenas os campos necessários.
• O comando KEYS * também é O(n) e bloqueia o servidor. Nunca o utilize em ambiente de produção. Para iteração sobre chaves, use a família de comandos SCAN (será detalhado em artigos futuros).

Porblemas Comuns e Soluções

• redis.exceptions.DataError: Ocorre ao tentar realizar operações numéricas (INCR, DECR, HINCRBY) em valores de String ou campos de Hash que não são números. Verifique sempre se o valor é numérico ou se a chave/campo não existe antes de operar.
• Valores de Campo em Hash: Lembre-se que todos os valores armazenados em campos de Hash são sempre strings. Se você armazena um número (ex: '30'), ao recuperá-lo, ele ainda será uma string. É responsabilidade da aplicação cliente converter para o tipo numérico adequado (int(), float()).
• HGETALL Retorna um Dicionário: Em redis-py, HGETALL retorna um dicionário Python, o que é conveniente. Verifique a documentação de outros clientes se estiver usando uma linguagem diferente.
• Crescimento Rápido da Memória:
  • Analise se Strings grandes estão sendo usadas indevidamente para armazenar objetos completos em vez de referências ou dados menores.
  • Verifique se Hashes estão com um número excessivo de campos; considere fragmentar o objeto em múltiplos Hashes se for o caso.

Exemplo Prático: Gerenciamento de Sessões de Usuário

Um caso de uso comum para Hashes é o armazenamento de dados de sessão de usuário, onde você precisa de acesso rápido a atributos específicos e a capacidade de atualizar a atividade de forma atômica.

import uuid
import time

class GerenciadorSessoes:
    def __init__(self, cliente_redis: Redis):
        self.r = cliente_redis
        self.tempo_expiracao_sessao = 30 * 60 # 30 minutos em segundos

    def criar_sessao(self, id_usuario: int, agente_navegador: str, **dados_extras) -> str:
        """Cria uma nova sessão de usuário usando um Hash para armazenar dados."""
        identificador_sessao = str(uuid.uuid4())
        chave_sessao = f'sessao:{identificador_sessao}'
        
        dados_sessao = {
            'id_usuario': str(id_usuario),
            'agente_navegador': agente_navegador,
            'criado_em': str(time.time()),
            'ultima_atividade': str(time.time()),
            **dados_extras
        }
        
        # Armazena os dados da sessão como um Hash e define um TTL
        self.r.hset(chave_sessao, mapping=dados_sessao)
        self.r.expire(chave_sessao, self.tempo_expiracao_sessao) # Define expiração
        
        return identificador_sessao

    def obter_sessao(self, identificador_sessao: str) -> dict | None:
        """Recupera informações de sessão (obtendo apenas os campos necessários, evitando HGETALL)."""
        chave_sessao = f'sessao:{identificador_sessao}'
        
        # Obtém apenas campos essenciais para verificar a existência da sessão
        id_usuario = self.r.hget(chave_sessao, 'id_usuario')
        if not id_usuario:
            return None # Sessão não existe ou expirou
        
        # Atualiza o timestamp da última atividade e reinicia o TTL
        self.r.hset(chave_sessao, 'ultima_atividade', str(time.time()))
        self.r.expire(chave_sessao, self.tempo_expiracao_sessao) 
        
        # Recupera outros campos conforme a necessidade da aplicação
        agente_navegador = self.r.hget(chave_sessao, 'agente_navegador')
        tema = self.r.hget(chave_sessao, 'tema') # Exemplo de campo extra
        
        return {'id_usuario': id_usuario, 'agente_navegador': agente_navegador, 'tema': tema}

    def atualizar_campo_sessao(self, identificador_sessao: str, campo: str, valor: str):
        """Atualiza um único campo da sessão (vantagem do Hash)."""
        chave_sessao = f'sessao:{identificador_sessao}'
        self.r.hset(chave_sessao, campo, valor)
        self.r.expire(chave_sessao, self.tempo_expiracao_sessao) # Refresca o TTL

# Exemplo de uso do Gerenciador de Sessões
gerenciador = GerenciadorSessoes(conexao_redis)
id_nova_sessao = gerenciador.criar_sessao(4001, 'Chrome/120.0', tema='escuro', idioma='en')
print(f"\nNova Sessão criada: {id_nova_sessao}")

dados_sessao_recuperados = gerenciador.obtener_sessao(id_nova_sessao)
print(f"Dados da Sessão: {dados_sessao_recuperados}")

gerenciador.atualizar_campo_sessao(id_nova_sessao, 'idioma', 'es')
print(f"Idioma da sessão atualizado. Novos dados: {gerenciador.obtener_sessao(id_nova_sessao)}")