Assinaturas compartilhadas MQTT: balanceamento de carga com $share

No MQTT tradicional, todo assinante cujo filtro corresponde a um tópico recebe sua própria cópia de cada mensagem. Isso é perfeito para fan-out, mas incômodo quando você quer que vários workers de backend dividam um fluxo de mensagens entre si. As assinaturas compartilhadas (shared subscriptions), padronizadas no MQTT 5, resolvem isso: os assinantes entram em um grupo nomeado, e o broker entrega cada mensagem a apenas um membro desse grupo.

A sintaxe $share

Uma assinatura compartilhada é uma solicitação de assinatura comum com um filtro de tópico especial:

$share/<ShareName>/<TopicFilter>

• $share — um prefixo literal que marca a assinatura como compartilhada.
• ShareName — o nome do grupo. Precisa ter pelo menos um caractere e não pode conter /, + ou #.
• TopicFilter — um filtro comum, com curingas inclusive. Veja tópicos e curingas MQTT.

Por exemplo, $share/ingest/factory/+/telemetry coloca o assinante no grupo ingest para todos os tópicos factory/<line>/telemetry. Os publicadores não fazem nada de especial: eles continuam publicando em factory/line-3/telemetry e nunca veem o prefixo $share. Para verificar quais tópicos um filtro abrange, cole a parte depois do nome do grupo no testador de tópicos MQTT.

Como as mensagens são distribuídas

As regras de entrega ficam mais claras com um exemplo. Suponha que existam estas assinaturas:

Ou seja, cada grupo recebe uma cópia, e dentro de um grupo um membro a processa. A especificação não define como esse membro é escolhido. Os brokers oferecem estratégias como round robin, aleatória, sticky (continua enviando ao mesmo membro até ele sair) ou hash do client ID ou do tópico — o EMQX, por exemplo, permite configurar isso. Cada membro mantém o QoS da sua própria assinatura, e as mensagens são confirmadas por membro, como de costume.

Suporte nos brokers

As assinaturas compartilhadas fazem parte do padrão MQTT 5, e um broker sinaliza o suporte por meio da propriedade Shared Subscription Available no CONNACK. Se um broker não as suporta, ele rejeita a assinatura com o reason code 0x9E (Shared Subscriptions not supported) — veja a referência de reason codes MQTT.

• O Mosquitto suporta $share desde a versão 1.6, para clientes MQTT 5 e 3.1.1.
• O EMQX suporta $share e também um atalho mais antigo, $queue/<filter>, que se comporta como uma assinatura compartilhada em um único grupo padrão.
• O HiveMQ suporta $share nas duas versões do protocolo.

Brokers gerenciados e em nuvem variam, então consulte a documentação antes de depender disso. Nossa comparação de brokers MQTT cobre as principais opções.

Exemplo: dois workers dividindo um fluxo

Abra três terminais. Os dois primeiros entram no mesmo grupo; o terceiro publica dez mensagens:

mosquitto_sub -V mqttv5 -h broker.hivemq.com -q 1 \
  -t '$share/workers/testmqtt/share-demo/jobs' -v

for i in $(seq 1 10); do
  mosquitto_pub -V mqttv5 -h broker.hivemq.com -q 1 \
    -t 'testmqtt/share-demo/jobs' -m "job $i"
done

Cada job aparece em exatamente um dos terminais dos workers. Repare nas aspas simples em volta do filtro: na maioria dos shells, $share entre aspas duplas seria expandido como variável. O mesmo funciona em código:

import mqtt from 'mqtt';

const client = mqtt.connect('wss://broker.hivemq.com:8884/mqtt', { protocolVersion: 5 });

client.on('connect', () => {
  client.subscribe('$share/workers/testmqtt/share-demo/jobs', { qos: 1 });
});

client.on('message', (topic, payload) => {
  // topic is the real topic, without the $share prefix
  console.log(topic, payload.toString());
});

Você pode ser o publicador pelo navegador: abra o cliente abaixo e publique em testmqtt/share-demo/jobs enquanto seus workers estão rodando.

Publicar jobs de teste no cliente online

Casos de uso

• Escalonamento horizontal de consumidores de backend. Rode N instâncias de um serviço de ingestão e deixe o broker distribuir a telemetria entre elas, sem uma fila de mensagens separada.
• Alta disponibilidade. Se um consumidor cair, os demais membros continuam recebendo mensagens.
• Integração com bancos de dados ou processadores de stream. Vários writers podem dividir a carga de inserir mensagens em um banco de séries temporais ou de encaminhá-las ao Kafka.
• Filas de trabalho. Comandos como "redimensionar esta imagem" podem ser tratados por qualquer worker que os receba.

Projetando uma configuração com assinaturas compartilhadas

Escolha nomes de grupo por tipo de consumidor

Dê aos grupos o nome da tarefa, não da instância: $share/ingest/..., $share/alerts/.... Todas as instâncias do serviço de ingestão entram em ingest; todas as instâncias do serviço de alertas entram em alerts. Assim, cada serviço como um todo vê cada mensagem exatamente uma vez, enquanto suas instâncias dividem o trabalho. Se você der sem querer um nome de grupo único para cada instância, volta ao fan-out simples e todas as instâncias processam todas as mensagens.

Sessões e QoS para os workers

Os workers normalmente assinam com QoS 1 para que uma mensagem seja reentregue se um worker cair antes de confirmá-la. Se as mensagens ficam enfileiradas para um grupo enquanto todos os membros estão offline depende das sessões dos membros: com uma sessão persistente, o broker pode continuar enfileirando; com clean sessions, mensagens publicadas enquanto ninguém do grupo está conectado simplesmente não são entregues a esse grupo.

Assinaturas compartilhadas vs uma fila de mensagens

As assinaturas compartilhadas oferecem consumidores concorrentes simples dentro do MQTT. Elas não oferecem recursos de sistemas de fila dedicados, como replay a partir de um offset, dead-letter queues ou garantias de ordenação por partição. Para muitos backends de IoT, esse equilíbrio é aceitável; se você precisa de replay, encaminhe o fluxo de uma assinatura compartilhada para um log como o Kafka e processe-o lá.

Armadilhas e limitações

• Sem ordenação global. O MQTT garante a ordem por tópico para um único assinante, mas quando as mensagens são distribuídas entre vários workers, elas são processadas em paralelo. Se a ordem importa por dispositivo, use uma estratégia do broker que faça hash do tópico ou particione por tópico você mesmo.
• Sem mensagens retidas. Pela especificação do MQTT 5, mensagens retidas não são enviadas quando uma assinatura compartilhada é estabelecida. Use uma assinatura normal para estado, como em tópicos de status retidos (retained).
• No Local não é permitido. Definir a opção No Local em uma assinatura compartilhada é um erro de protocolo no MQTT 5.
• Cancele a assinatura com o filtro completo. Para sair do grupo, cancele a assinatura usando exatamente a string $share/group/filter, não apenas o filtro simples.
• ACLs. Os brokers diferem quanto a aplicar as regras de acesso à string $share/... ou ao filtro subjacente. Teste suas permissões com a sintaxe real.
• Duplicatas continuam possíveis. QoS 1 significa pelo menos uma vez, e a reentrega a outro membro após uma desconexão pode gerar duplicatas. Projete handlers idempotentes; veja níveis de QoS do MQTT.

Perguntas frequentes