Patrick Brandão

Redis – PubSub

Saudações.

Esse tutorial vai abordar o recurso PubSub do Redis e Valkey de comunicação em tempo real entre sistemas.

Redis e Valkey operam o protocolo RESP2 e RESP3. Vou explicar as diferenças no protocolo, ambos os softwares são gêmeos.

1 – O que é PubSub

Vou explicar como funciona o mecanismo de publicação (Pub) e assinatura (Sub) que nasceu no Redis e vive também no Valkey.

O PubSub é a base de sistemas de mensageria, nesse ecosistema, softwares conversam entre si sem se conectarem uns aos outros.

1.1 – O que existia antes do PubSub

Criar um único software que faz tudo e roda num único processo cheio de threads e processos filhos clonados é complexo, lento e sujeito a quedas catastróficas, alem de ser difícil de escalar horizontalmente.

Métodos tradicionais (herança do SOA) eram utilizados para separar os softwares em partes que se comunicam como operários em uma linha de montagem.

Formas que os processos usavam para comunicação:

  • Alvo fixo: O software A sabe o endereço do software B;
    • Rede IP: IP (10.90.2.14), porta (9281), protocolo (UDP, TCP, HTTP), na maioria dos casos por URLs (https://api.dominio.com/v1/auth/check/user);
    • Socket Unix: Por canal interno no sistema operacional (/var/run/docker.sock);
    • File System: A criação de um arquivo sinaliza ao software leitura que há trabalho a ser feito, o arquivo é removido (e as vezes entregue pronto em outro diretório);
  • Ponto a ponto: Toda conversa entre softwares requer como pré-requisito saber o endereço do recurso (URL), o que acaba resultando em um ambiente mesh onde um monte de chamadas ocorrem para um montão de URLs em paralelo;
  • Pooling: O software fica verificando em loop se um novo trabalho está disponível, se ficou pronto, se há updates, etc… isso resulta em dois problemas:
    • Latência: Se o loop possui pausa de 100ms, aplicações que precisam de resposta em tempo real podem apresentar lentidão;
    • Sobrecarga: Se reduzir a pause do loop o cliente e o servidor consumirão mais CPU, memória, rede, I/O, aumentando os custos computacionais, se muitos softwares fazem isso apontando para um único servidor isso resulta em catástrofe.

1.2 – Criação do PubSub

O principal problema da arquitetura SOA era a necessidade de pooling para reagir a eventos.

Evitamos o pooling ao usar protocolos que podem manter a conexão ativa sem tráfego e sem precisar enviar requests de verificação. Devemos edisparar eventos somente quando uma nova mensagem chegar, temos alguns protocolos como soluções:

  • HTTP com SSE (Streamable HTTP);
  • WebSocket;
  • RESP;

O RESP é onde o Redis e Valkey atuam.

O PubSub cria uma arquitetura HUB-AND-SPOKE: Os software combinam um canal único passivo em tempo real, faz-se uma única conexão (TCP/IP), todos os clientes irão se conectar e trocar dados por ela, o servidor atuará como o HUB, que nesse caso recebe o nome específico de broker.

Algum cliente conectado ao broker deseja enviar uma mensagem, ele se torna um publicador (publisher), e publica (publish) a mensagem em um canal (channel), e todos os clientes que estão conectados no mesmo canal são assinantes (subscribers), que recebem a mensagem instantaneamente.

O PubSub do RESP é um broadcast, todos os assinantes do canal recebem a mesma cópia.

Esse desacoplamento total entre quem envia e quem recebe é o coração do padrão publish/subscribe e é o que torna esse mecanismo tão útil para notificações ao vivo, salas de chat, atualização de painéis e comunicação entre microsserviços.

1.3 – Canal agnóstico

Do ponto de vista do canal, o publicador não sabe a identidade do cliente que está ouvindo, e o ouvinte não sabe a identidade de quem publicou. Não há como saber o nome do software, seu IP e porta de origem, NADA é informado, somente o payload puro da mensagem.

Se você não gosta desse anonimato nativo do canal, basta que a mensagem tenha um formato de envelope contendo remetente e detalhes técnicos alem do payload objetivo.

O canal é agnóstico ao conteúdo da mensagem, ele não implementa isso no RESP, você, engenheiro de software, deve definir o formato e os metadados desse envelope.

1.4 – Comandos PubSub

Fluxo de uso:

  • CONNECTION: O software se conecta ao software que faz o papel de HUB;
  • SUBSCRIBE: O software se inscreve em um canal, se o canal não existir ele passa a existir com 1 assinante;
  • PSUBSCRIBE: O software se inscreve em vários canais usando coringa, o nome dos canais que passar no teste coringa serão monitorados;
  • PUBLISH: O software envia uma mensagem no canal:
    • Retorno 0: Canal sem assinantes (ninguem pra ouvir)
    • Retorno >=1: Existem assinantes, informa quantos receberam;
  • UNSUBSCRIBE: O software deseja deixar o canal;
  • PUNSUBSCRIBE: O software deseja deixar vários canais usando um coringa.

1.5 – Canal sem compromisso

O PubSub não garante a entrega, não persiste dados, não repete mensagens, ele atua na lógica “at-most-once” (no máximo uma vez), explicando:

  • Quem está inscrito no momento do envio recebe uma cópia, quem não está inscrito não recebe (obviamente) e nem saberá que a mensagem existiu ao se inscrever após esse envio;
  • Se um inscrito se desconectar e reconectar, e nos 1ms em que ele ficou fora alguma mensagem foi enviada no canal, ele perde a mensagem (nem saberá que existiu);
  • O produtor sabe o número de inscritos que receberam a mensagem, se não há ninguém no canal, ele pode, por decisão própria do software (não do canal), repetir mais tarde o envio. Precisa ficar claro que isso não é feito pelo PubSub e sim pelo produtor por conta própria (você tem que escrever essa lógica);
  • Não ha, no canal, recursos para consultar histórico e obter mensagens anteriores, mais uma vez isso pode ser implementado por você combinando técnicas de PubSub com listas.

Agora que ficou claro as regras básicas e simples, vamos ao laboratório prático.

2 – Testando o PubSub

Vou criar um container do Valkey para os testes.

2.1 – Rede Docker

Crie a rede “network_public“:

Bash
# Rede de containers
docker network create network_public \
    -d bridge \
    -o com.docker.network.bridge.name=br-net-public \
    -o com.docker.network.driver.mtu=1500 \
    -o com.docker.network.bridge.gateway_mode_ipv4=nat-unprotected \
    --subnet 10.249.0.0/16 \
    --gateway 10.249.255.254;

2.2 – Container Valkey

Vamos criar o container chamado pubsub-server (sem persistência em disco):

Bash
# Variaveis
    NAME="pubsub-server";

# Imagem (escolha valkey ou redis)
    IMAGE="valkey/valkey:9.1-alpine";
    #IMAGE="redis:8.8-alpine";

# Renovar/rodar:
    docker container rm -f $NAME 2>/dev/null;
    docker container run \
        --detach    \
        --read-only \
        --name                $NAME \
        --hostname            $NAME.intranet.br \
        --network             network_public \
        --ip                  10.249.199.199 \
        --restart             always \
        $IMAGE \
            --save "" \
            --appendonly no \
            --cluster-enabled no \
            --replica-read-only no;

2.3 – Conectando como cliente

Abra dois terminais, em cada um entre container pubsub-server usando o cliente RESP com o comando redis-cli (Valkey e Redis tem esse mesmo comando):

Bash
# Acessando redis-cli no container pubsub-server:
docker exec -it pubsub-server redis-cli;

No terminal 1:

Bash
# Acessando redis-cli no container pubsub-server:
docker exec -it pubsub-server redis-cli;
    # 127.0.0.1:6379>

No terminal 2:

Bash
# Acessando redis-cli no container pubsub-server:
docker exec -it pubsub-server redis-cli;
    # 127.0.0.1:6379>

2.4 – SUBSCRIBE e PUBLISH

Agora vamos escutar mensagens, para isso iremos fazer inscrição (subscribe) em dois canais: “noticias” e “alertas“. Use o terminal 1.

Você pode se inscrever em vários canais se preferir, isso ajuda a economizar conexões TCP entre os softwares clientes e o servidor.

Bash
SUBSCRIBE noticias alertas
    # 1) "subscribe"
    #    2) "noticias"
    #    3) (integer) 1
    #
    # 1) "subscribe"
    #    2) "alertas"
    #    3) (integer) 2
    #

# 127.0.0.1:6379(subscribed mode)> _

E agora vamos mandar mensagens nos canais usando o terminal 2:

Bash
# Enviar mensagem no canal 'noticias'
PUBLISH  noticias  "O rato roeu a roupa do rei de roma"
    # (integer) 1

# Enviar mensagem no canal 'alertas'
PUBLISH  alertas  "Exterminador de ratos nao fez um bom trabalho"
    # (integer) 1

Obs: O protocolo RESP3 permite que outros comandos (SET, GET, …) sejam executados enquanto você está inscrito nos canais, tecle ENTER para obter a linha de comando, tecle ENTER sem nenhum comando para mostrar as mensagens recebidas enquanto você usava o shell.

Resultado no terminal 1:

Bash
127.0.0.1:6379(subscribed mode)> 
    1) "message"
    2) "noticias"
    3) "O rato roeu a roupa do rei de roma"

    1) "message"
    2) "alertas"
    3) "Exterminador de ratos nao fez um bom trabalho"

Reading messages... (press Ctrl-C to quit or any key to type command)

3 – PubSub no Cluster

Uma base de mensageria que precise escalar para milhões de operações por segundo como logs e telemetria precisará ser distribuida por vários nós em cluster.

3.1 – Particionamento

O RESP lida com cluster por meio de shardding (particionamento).

Cada servidor fica responsável por um número de chaves (slot), o hash do nome da chave determina a qual slot ele pertence e por consequência define em qual nó do cluster ela mora.

Quando mais nós são adicionados no cluster maior a capacidade de operações em tempo real.

Infelizmente ao usar os comandos SUBSCRIBE e PUBLISH isso não acontece. O canal gerado com esses comandos integram todos os nós, uma mensagem enviado para o primeiro nó será retransmitida a todos os demais gerando uma inundação. Os clientes podem se conectar em qualquer nó que mensagens serão recebidas – facilidade para o cliente, stress para o cluster.

3.2 – Canal na partição

Para resolver isso existem os comandos SSUBSCRIBE e SPUBLISH (S=sharded), criados especificamente para uso no Cluster. Un canal criado no modo sharded fica isolado no servidor a qual seu slot pertence, seguindo a mesma lógicas das demais chaves clusterizadas.

Inscrição particionada (sharded subscribe) e a publicação particionada (sharded publish):

Bash
# Terminal 1
# Inscrevendo - sharded subscribe
SSUBSCRIBE pedidos

# Terminal 2
# Publicando - sharded subscribe
SPUBLISH pedidos "novo pedido #1001"

# Saida no Terminal 1
    # 1) "ssubscribe"
    # 2) "pedidos"
    # 3) (integer) 1
    # 1) "smessage"
    # 2) "pedidos"
    # 3) "novo pedido #1001"

# Informacoes dos canais particionados
PUBSUB SHARDCHANNELS
PUBSUB SHARDNUMSUB pedidos

Resumo rápido: em ambiente de cluster com alto volume, use a família S* (sharded). Tanto o método não-particionado quando o particionado funcionam no Redis de instância única.

4 – Problemas e soluções

Vou descrever alguns casos que apresentam problemas e soluções diferentes.

4.1 – Contexto de assinante

Ao tentar utilizar uma conexão como inscrito (SUBSCRIBE) você estará por padrão usando o protocolo RESP2. Se tentar enviar mensagens como publicador (PUBLISH) você enfrentará um erro. Observe:

Bash
# Terminal 1 - apenas inscrito
SUBSCRIBE noticias

# Terminal 2 - inscrever no canal 'noticias' e depois enviar uma mensagem nele
SUBSCRIBE noticias
PUBLISH noticias "o rato roeu a roupa do rei de roma"
    # ERRO 
    # (error) ERR Can't execute 'publish':
    #    only
    #        (P|S)SUBSCRIBE /
    #        (P|S)UNSUBSCRIBE /
    #        PING / QUIT / RESET
    # are allowed in this context

Quando uma conexão executa SUBSCRIBE, PSUBSCRIBE ou SSUBSCRIBE, ela entra no que a documentação chama de subscriber context (modo assinante).

A partir daí, aquela conexão passa a receber mensagens push assíncronas do servidor a qualquer momento.

No RESP2 não existe forma de distinguir, no fluxo de bytes, uma resposta a um comando de uma mensagem publicada pois os dois chegam como arrays comuns.

Para evitar essa ambiguidade, o servidor bloqueia comandos normais enquanto a conexão está nesse modo.

Os únicos comandos permitidos em modo assinante (RESP2) são:

  • SUBSCRIBE,
  • UNSUBSCRIBE,
  • PSUBSCRIBE,
  • PUNSUBSCRIBE,
  • SSUBSCRIBE,
  • SUNSUBSCRIBE,
  • PING,
  • QUIT e
  • RESET

O comando PUBLISH não está na lista, por isso o erro.

Solução: fazer upgrade para RESP3, que permite que a mesma sessão seja usada para várias operações paralelas sem misturar resultados.

4.2 – Usando RESP3

Ao usar RESP3 você afeta as capacidade do cliente de multiplexar conteudo, não afeta os demais clientes conectado no mesmo servidor PubSub, ou seja, você pode ser o único terminal na versão 3 enviando e recebendo simultaneamente enquanto todos os demais clientes no mesmo servidor podem estar usando versão 2 sem prejuizo das funcionalidades de somente consumir ou somente produzir.

No terminal 1:

Bash
# Acessando redis-cli no container pubsub-server usando RESP2 (legado)
docker exec -it pubsub-server redis-cli -2;
    # 127.0.0.1:6379>

No terminal 2:

Bash
# Acessando redis-cli no container pubsub-server usando RESP3 (comando HELLO, novo):
docker exec -it pubsub-server redis-cli -3;
    # 127.0.0.1:6379>

Testando o full duplex na mesma conexão:

Bash
# Terminal 1 - apenas inscrito
SUBSCRIBE noticias

# Terminal 2 - inscrever no canal 'noticias' e depois enviar uma mensagem nele
SUBSCRIBE noticias
PUBLISH noticias "o rato roeu a roupa do rei de roma"
    # 1) "message"
    # 2) "noticias"
    # 3) "o rato roeu a roupa do rei de roma"

Funciona!

4.3 – Buffers e Payload

O PubSub é um ótimo recurso para implementar rapidamente salas de Chat online, a arquitetura é simples:

  • Servidor PubSub: O container pubsub-server no nosso caso;
  • Worker: Software desenvolvido para fazer ponte entre os clientes na Internet e o PubSub:
    • Servidor HTTP/WebSocket: provê a comunicação entre o navegador do usuário e o servidor;
    • Cliente RESP: faz a comunicação com o servidor PubSub;
    • O que vem do usuário é enviado para o canal;
    • O que vem do canal é enviado para o usuário;
    • O nome da sala de chat é o nome do canal.

Se o APP de Chat crescer em usuários e sessões, aumenta-se a quantidade de workers, 10 deles atendendo 100.000 usuários, é fácil de implementar e consome poucos recursos.

Todos os workers deveriam, no mundo ideal, estarem no mesmo datacenter com alta largura de banda e baixa latência até o servidor PubSub.

Infelizmente esaa não é a realidade. Ambientes de Cluster Docker Swarm e Kubernetes acabam por espalhar os containers por vários nós em locais diferente e até países diferentes (cluster feito com VPS baratinha).

Aqui surge o problema dos buffers. Assinantes lentos derrubam a festa. Cada conexão tem um buffer de saída.

Se o publicador despeja mensagens mais rápido do que o assinante consegue consumir, esse buffer cresce. No algoritmo do balde furado, a torneira enche o balde mais rápido do que o furo consegue se livrar da água, uma hora o balde irá entornar.

Ao estourar o limite configurado no parâmetro client-output-buffer-limit do servidor (que tem um perfil específico para clientes pubsub), o servidor desconecta o assinante para se proteger.

O sintoma é um assinante que “cai sozinho” sob carga elevada, os workers acusam erros de desconexão.

As soluções são:

  • Melhorar o worker: Tornar o consumidor mais rápido melhorando a rede, dando mais CPU/RAM, colocando o worker mais perto do servidor;
  • Payload reduzido: Reduzir o volume publicado, enviar mensagens com payload menor;
    • Em vez de transmitir mensagens grandes no canal, coloque o payload grande em uma chave cujo nome é um UUID aleatório (UUIDv4, UUIDv7) e transmita o nome da chave no canal;
    • Se o canal e o Redis não for adequado para o volume, pense em hospedá-lo em S3, NoSQL ou SQL e informe o caminho para o recurso no canal;
    • Todos os workers passam a receber as publicações em tempo real (poucos bytes por mensagem);
  • Aumentar o buffer: Não é a melhor ideia, prejudicará o canal inteiro, mas pode amenizar temporariamente, comandos:
    • CONFIG GET client-output-buffer-limit
    • CONFIG SET client-output-buffer-limit "pubsub 64mb 16mb 6

4.4 – Competing Consumers

A técnica de consumidores concorrentes é usada para combinar trabalhos que só um worker pode assumir, mesmo que todos sejam comunicados.

Nela o “trabalho” é inserido em um banco de dados ACID como o PostgreSQL, o registro é inserido sem um dono e o ID do registro é comunicado no canal.

O primeiro worker que conseguir fazer a reserva do registro por meio da transação que preenche seu nome de worker responsável fica com o registro para si.

Essa técnica é adequada apenas em ambientes pequenos, onde o PubSub dispara eventos em tempo real mas também permite que os workers façam pooling na tabela para continuar os trabalhos acumulados anteriormente, suporta trabalhos progressivos que caso falhem em 50%, o worker continua nos 50% quanto retomar a tarefa.

É especialmente aplicada em ambientes onde nem todos os softwares podem participar da mensageria.

SQL
-- Tabela - UUIDv7 requer Postgres 18
CREATE TABLE jobs (
    job_uuid         uuid         PRIMARY KEY DEFAULT uuidv7(),
    worker_name      text         NOT NULL DEFAULT '',
    job_content      jsonb        NOT NULL DEFAULT '{}'::jsonb,
    created_at        timestamptz  NOT NULL DEFAULT NOW(),
    updated_at        timestamptz  NOT NULL DEFAULT NOW(),
    blocked           smallint     NOT NULL DEFAULT 0,
);

-- Inserir novo job
INSERT INTO jobs (job_content) VALUES ('{id:1983,price:183,type:"cart"');
-- (obter "job_uuid" gerado e enviar no canal PUBSUB

-- Cada Worker
-- Transacao para reserva da tarefa por um worker,
-- Todos os workers vao executar o mesmo bloco abaixo com seus nomes (worker_name),
-- Somente 1 worker consegue assumir
BEGIN;
    UPDATE
        jobs
            SET
                blocked = 1,
                worker_name="my_worker_unique_name"
            WHERE
                job_uuid = "<uuid-recebido-no-canal-pub-sub>" AND
                blocked = 0 AND
                worker_name = ""
            LIMIT 1;
COMMIT;

-- Worker verifica se conseguiu assumir a tarefa apos executar a transacao
-- de competicao acima
SELECT *
    FROM jobs
        WHERE
                job_uuid = "<uuid-recebido-no-canal-pub-sub>" AND
                blocked = 1 AND
                worker_name="my_worker_unique_name";

4.5 – Canal de eventos internos

Algumas soluções requerem monitoramento de chaves, para não estressar o servidor com busca de chaves (algo extremamente bloqueante), é possível ativar as notificações de alterações de chaves.

Ao fazer isso o servidor irá criar canais alimentados internamente com as modificações de chaves (keyspace events).

Exemplo:

Bash
# Ativar Key Events
CONFIG SET notify-keyspace-events KEA

# Monitorar canais de eventos internos
PSUBSCRIBE __keyevent@0__:*

# Criar chaves para testar recurso
SET  nome  Patolino
DEL  nome

Terminamos por hoje!

Patrick Brandão, patrickbrandao@gmail.com

“Em algum lugar, alguma coisa incrível
está esperando para ser descoberta.“
Carl Sagan