Assinaturas Compartilhadas MQTT: Balanceamento de Carga entre Consumidores

As assinaturas compartilhadas MQTT (shared subscriptions) permitem que vários clientes se inscrevam como um grupo, de modo que cada mensagem de um topic seja entregue a apenas um membro desse grupo, e não a todos. Ao final deste artigo você vai entender a sintaxe $share/group/topic, como a entrega funciona quando workers entram e saem, como escalar consumidores de backend com esse recurso e como verificar se o broker que você usa oferece suporte a ele.

Por que assinaturas comuns não escalam consumidores

No publish/subscribe padrão, todo assinante cujo filtro corresponde recebe sua própria cópia de cada mensagem. É exatamente o que você quer para dashboards: cada tela mostra cada leitura. E é exatamente o que você não quer no processamento de backend. Se você executa três instâncias de um serviço que grava a telemetria de uma mineradora em um banco de dados, três assinaturas comuns significam que cada leitura é gravada três vezes.

Antes do MQTT 5, as alternativas eram desajeitadas: um único consumidor que virava gargalo, particionar topics manualmente entre instâncias ou fazer uma ponte do MQTT para um sistema de filas separado. As assinaturas compartilhadas resolvem o problema dentro do próprio broker MQTT.

Como funcionam as assinaturas compartilhadas MQTT

Uma assinatura compartilhada usa um formato especial de filtro, padronizado no MQTT 5:

$share/<ShareName>/<TopicFilter>

• $share é o prefixo fixo que marca a assinatura como compartilhada.
• <ShareName> dá nome ao grupo de consumidores. Não pode conter /, + ou #.
• <TopicFilter> é um filtro normal, com curingas inclusive.

Todo cliente que se inscreve no mesmo $share/<ShareName>/<TopicFilter> entra no mesmo grupo. Para cada mensagem correspondente, o broker escolhe um membro do grupo e entrega a mensagem a ele. Nomes de compartilhamento diferentes são grupos independentes, e assinantes comuns do topic simples continuam recebendo suas próprias cópias.

A especificação deixa a escolha do membro a critério do broker. Estratégias round-robin, aleatória e "menos ocupado" são todas comuns, então não conte com uma ordem específica nem com mensagens de um dispositivo chegando sempre ao mesmo worker.

Semântica de entrega que você precisa conhecer

1. A ordenação é por membro, não por topic. Cada worker vê as próprias mensagens em ordem, mas duas leituras consecutivas do mesmo sensor podem ir para workers diferentes e ser processadas em paralelo. Se a ordem importa, inclua um número de sequência ou timestamp e resolva isso no banco de dados.
2. O QoS continua valendo. Com QoS 1, se um worker se desconecta antes de confirmar uma mensagem, os brokers normalmente a reentregam à mesma sessão quando ele se reconecta, ou a encaminham para outro membro; o comportamento varia conforme o broker. Torne seus handlers idempotentes.
3. Retained messages não são enviadas na inscrição. A especificação diz que retained messages não são entregues a uma assinatura compartilhada quando ela é criada, porque não ficaria claro qual membro deveria recebê-las.
4. No Local é proibido. Definir a opção de assinatura No Local em uma assinatura compartilhada é um erro de protocolo.
5. As sessões são por membro. Cada worker tem seu próprio client ID e sua própria sessão. Uma sessão persistente em um worker mantém a assinatura dele ativa enquanto ele está offline. Veja sessões persistentes MQTT para entender como isso afeta o enfileiramento.

Exemplo: escalando workers em Python

O worker abaixo entra em um grupo chamado db-writers e processa telemetria. Execute-o em dois ou três terminais com números de worker diferentes para ver a carga se distribuir.

import sys
import json
import paho.mqtt.client as mqtt

WORKER = sys.argv[1] if len(sys.argv) > 1 else "1"
FILTER = "$share/db-writers/connectmq/demo/s7h3/factory/+/telemetry"

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"Worker {WORKER} connected: {reason_code}")
    client.subscribe(FILTER, qos=1)

def on_subscribe(client, userdata, mid, reason_code_list, properties):
    for rc in reason_code_list:
        if rc.is_failure:
            print(f"Subscribe rejected: {rc}")  # ex.: Shared Subscriptions not supported
        else:
            print(f"Joined group, granted {rc}")

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    print(f"Worker {WORKER} got {msg.topic}: {data}")
    # grave no seu banco de dados aqui; mantenha a operação idempotente

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"db-writer-s7h3-{WORKER}",
                     protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_subscribe = on_subscribe
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Inicie dois workers:

python worker.py 1
python worker.py 2

Em seguida, publique uma sequência de leituras de teste:

for i in $(seq 1 10); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -i sh-pub-s7h3 -q 1 \
    -t "connectmq/demo/s7h3/factory/line$((i % 3))/telemetry" \
    -m "{\"seq\": $i, \"temp\": 70}"
done

Se o broker oferecer suporte a assinaturas compartilhadas, as dez mensagens são divididas entre os workers, em vez de cada worker imprimir todas as dez. Substitua s7h3 por uma string aleatória sua; o servidor MQTT público (gratuito) serve para testes e protótipos e pode ser lido por qualquer pessoa, então publique apenas dados fictícios.

A mesma assinatura em MQTT.js fica assim:

const mqtt = require("mqtt");
const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId: "db-writer-node-s7h3-1",
  protocolVersion: 5,
});
client.on("connect", () => {
  client.subscribe("$share/db-writers/connectmq/demo/s7h3/factory/+/telemetry",
    { qos: 1 }, (err, granted) => console.log(err || granted));
});
client.on("message", (topic, payload) => console.log(topic, payload.toString()));

Como verificar o suporte a assinaturas compartilhadas

O suporte varia entre brokers e entre planos, então verifique em vez de presumir:

1. Leia o SUBACK. No MQTT 5, um broker que não suporta assinaturas compartilhadas retorna o reason code 0x9E (Shared Subscriptions not supported). O callback on_subscribe acima o imprime.
2. Confira o CONNACK. Brokers MQTT 5 podem anunciar Shared Subscription Available = 0 no CONNACK. Se a propriedade estiver ausente, presume-se que há suporte.
3. Faça o teste de divisão. Dois membros, dez mensagens. Se os dois membros imprimirem todas as dez, o broker tratou $share/... como um filtro de topic comum, e não como um grupo.
4. Verifique o plano. Brokers MQTT gerenciados costumam vincular o recurso a um nível de plano. No ConnectMQ, as assinaturas compartilhadas estão incluídas no plano Serverless (Paid) e superiores.
5. Teste com MQTT 3.1.1 separadamente. Alguns brokers aceitam $share de clientes 3.1.1 como extensão; outros exigem MQTT 5. Não conte com isso a menos que o seu broker documente.

Assinaturas compartilhadas comparadas a outras abordagens

Existem outras formas de dividir o trabalho, e vale saber quando cada uma se encaixa. O particionamento manual de topics (o worker 1 se inscreve em line1, o worker 2 em line2) garante ordenação estrita por partição, mas exige reconfiguração sempre que você adiciona um worker ou uma linha. Um único consumidor alimentando uma fila interna mantém a ordenação simples, mas transforma esse processo em gargalo e ponto único de falha. As assinaturas compartilhadas não exigem coordenação entre workers e escalam simplesmente iniciando mais instâncias, ao custo da ordenação por topic entre workers. Para a maior parte da ingestão de telemetria e do processamento de alertas, essa troca é a certa.

Padrões e armadilhas

• Um grupo por finalidade. Use um nome de compartilhamento por tarefa independente (db-writers, alerting, exporter). Assim, cada tarefa recebe cada mensagem exatamente uma vez entre seus workers.
• Autoscaling. Adicionar um worker é simplesmente conectar outro cliente com o mesmo filtro; remover um é desconectá-lo. Dê a cada instância um client ID único, como explicado em boas práticas de client ID MQTT.
• Controle de fluxo. Combine assinaturas compartilhadas com um Receive Maximum pequeno do MQTT 5 em cada worker, para que uma instância lenta não receba uma grande rajada de mensagens não confirmadas.
• Curingas funcionam bem. $share/group/factory/# distribui um ramo inteiro. As regras de correspondência são as de sempre, explicadas em curingas MQTT.
• Não use para fan-out. Se todo consumidor precisa ver cada mensagem, use assinaturas comuns.

Está começando com MQTT? O tutorial de MQTT cobre assinaturas, QoS e sessões desde o início.

Escale consumidores no ConnectMQ

O ConnectMQ é um broker MQTT gerenciado com suporte a MQTT 3.1.1 e 5.0, com assinaturas compartilhadas disponíveis a partir do plano Serverless (Paid), além de QoS 2, mensagens de 1 MB e até 1.000 conexões. As opções Dedicated e BYOC estão disponíveis quando você precisa de um cluster isolado. Cadastre-se e comece no plano Serverless gratuito sem cartão de crédito e faça o upgrade quando o seu backend precisar de grupos de consumidores. Conecte seus workers com YOUR_USERNAME, YOUR_PASSWORD e YOUR_CLIENT_ID.

Perguntas frequentes

Qual é a sintaxe de uma assinatura compartilhada MQTT?

Use $share/<ShareName>/<TopicFilter>, por exemplo $share/workers/sensors/+/temperature. Todos os clientes que usam o mesmo nome de compartilhamento e o mesmo filtro formam um grupo.

Assinaturas compartilhadas funcionam com MQTT 3.1.1?

Elas são definidas no MQTT 5. Alguns brokers aceitam a sintaxe $share de clientes 3.1.1 como extensão, mas você deve confirmar isso na documentação do seu broker antes de depender disso.

Retained messages são entregues a assinaturas compartilhadas?

Não. Quando uma assinatura compartilhada é criada, as retained messages não são enviadas a ela. Leia o estado retido com uma assinatura comum separada se um worker precisar dele.

Como o broker escolhe qual assinante recebe uma mensagem?

A especificação MQTT deixa isso a cargo do broker. Seleção round-robin e aleatória são comuns, então nunca dependa de um membro específico receber uma mensagem específica.