Abonnements partagés MQTT : répartir la charge entre consommateurs

Les abonnements partagés MQTT permettent à plusieurs clients de s'abonner en groupe, de sorte que chaque message d'un topic soit livré à un seul membre de ce groupe plutôt qu'à tous. À la fin de cet article, vous comprendrez la syntaxe $share/group/topic, le fonctionnement de la livraison lorsque des workers arrivent et repartent, la manière de faire monter en charge vos consommateurs backend et comment vérifier si votre broker prend en charge cette fonctionnalité.

Pourquoi les abonnements classiques ne font pas monter les consommateurs en charge

En publish/subscribe standard, chaque abonné dont le filtre correspond reçoit sa propre copie de chaque message. C'est exactement ce que l'on veut pour les tableaux de bord : chaque écran affiche chaque mesure. C'est exactement ce que l'on ne veut pas pour le traitement backend. Si vous exécutez trois instances d'un service qui écrit la télémétrie dans une base de données, trois abonnements classiques signifient que chaque mesure est écrite trois fois.

Avant MQTT 5, les contournements étaient maladroits : un consommateur unique qui devenait un goulet d'étranglement, un partitionnement manuel des topics entre les instances, ou un pont entre MQTT et un système de files d'attente séparé. Les abonnements partagés résolvent le problème au sein même du broker MQTT.

Comment fonctionnent les abonnements partagés MQTT

Un abonnement partagé utilise un format de filtre particulier, standardisé dans MQTT 5 :

$share/<ShareName>/<TopicFilter>

• $share est le préfixe fixe qui marque l'abonnement comme partagé.
• <ShareName> nomme le groupe de consommateurs. Il ne doit contenir ni /, ni +, ni #.
• <TopicFilter> est un filtre classique, wildcards compris.

Chaque client qui s'abonne au même $share/<ShareName>/<TopicFilter> rejoint le même groupe. Pour chaque message correspondant, le broker choisit un membre du groupe et lui livre le message. Des noms de partage différents forment des groupes indépendants, et les abonnés classiques au topic simple reçoivent toujours leurs propres copies.

La spécification laisse au broker le choix du membre. Les stratégies round-robin, aléatoire et « le moins occupé » sont toutes courantes : ne comptez donc ni sur un ordre précis, ni sur le fait que les messages d'un même appareil atteignent toujours le même worker.

La sémantique de livraison à connaître

1. L'ordre est garanti par membre, pas par topic. Chaque worker voit ses propres messages dans l'ordre, mais deux mesures consécutives d'un même capteur peuvent aller à des workers différents et être traitées en parallèle. Si l'ordre compte, incluez un numéro de séquence ou un horodatage et résolvez-le dans votre base de données.
2. La QoS s'applique toujours. En QoS 1, si un worker se déconnecte avant d'acquitter un message, les brokers le relivrent généralement à la même session lors de sa reconnexion, ou le routent vers un autre membre ; le comportement varie selon le broker. Rendez vos handlers idempotents.
3. Les retained messages ne sont pas envoyés à l'abonnement. La spécification précise que les retained messages ne sont pas livrés à un abonnement partagé lors de sa création, car on ne saurait pas quel membre devrait les recevoir.
4. No Local est interdit. Activer l'option d'abonnement No Local sur un abonnement partagé est une erreur de protocole.
5. Les sessions sont propres à chaque membre. Chaque worker a son propre client ID et sa propre session. Une session persistante sur un worker maintient l'abonnement de ce worker pendant qu'il est hors ligne. Consultez les sessions persistantes MQTT pour voir l'effet sur la mise en file.

Exemple : faire monter en charge des workers Python

Le worker ci-dessous rejoint un groupe nommé db-writers et traite la télémétrie. Lancez-le dans deux ou trois terminaux avec des numéros de worker différents pour observer la répartition de la charge.

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}")
    # écrivez dans votre base de données ici ; gardez l'opération 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()

Démarrez deux workers :

python worker.py 1
python worker.py 2

Publiez ensuite un flux de mesures de test :

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

Si le broker prend en charge les abonnements partagés, les dix messages sont répartis entre les workers au lieu que chacun les affiche tous les dix. Remplacez s7h3 par votre propre chaîne aléatoire ; le serveur MQTT public gratuit est lisible par tous et réservé aux tests : ne publiez donc que des données factices.

Le même abonnement avec MQTT.js ressemble à ceci :

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()));

Comment vérifier la prise en charge des abonnements partagés

La prise en charge varie selon les brokers et selon les offres : vérifiez plutôt que de supposer.

1. Lisez le SUBACK. En MQTT 5, un broker qui ne prend pas en charge les abonnements partagés renvoie le reason code 0x9E (Shared Subscriptions not supported). Le callback on_subscribe ci-dessus l'affiche.
2. Vérifiez le CONNACK. Les brokers MQTT 5 peuvent annoncer Shared Subscription Available = 0 dans le CONNACK. Si la propriété est absente, la prise en charge est présumée.
3. Faites le test de répartition. Deux membres, dix messages. Si les deux membres affichent les dix, le broker a traité $share/... comme un filtre de topic ordinaire et non comme un groupe.
4. Vérifiez l'offre. Les brokers MQTT managés lient souvent la fonctionnalité à un niveau d'offre. Sur ConnectMQ, les abonnements partagés sont inclus à partir de l'offre Serverless (Paid).
5. Testez MQTT 3.1.1 séparément. Certains brokers acceptent $share de la part de clients 3.1.1 en tant qu'extension ; d'autres exigent MQTT 5. Ne comptez pas dessus sauf si votre broker le documente.

Les abonnements partagés comparés aux autres approches

Il existe d'autres façons de répartir le travail, et il est utile de savoir quand chacune convient. Le partitionnement manuel des topics (le worker 1 s'abonne à line1, le worker 2 à line2) offre un ordre strict par partition, mais impose une reconfiguration à chaque ajout de worker ou de ligne. Un consommateur unique alimentant une file interne garde l'ordre simple, mais fait de ce processus un goulet d'étranglement et un point de défaillance unique. Les abonnements partagés ne nécessitent aucune coordination entre workers et montent en charge en démarrant simplement davantage d'instances, au prix de l'ordre par topic entre workers. Pour la plupart des cas d'ingestion de télémétrie et de traitement d'alertes, c'est le bon compromis.

Modèles et pièges

• Un groupe par finalité. Utilisez un nom de partage par tâche indépendante (db-writers, alerting, exporter). Chaque tâche reçoit alors chaque message exactement une fois pour l'ensemble de ses workers.
• Autoscaling. Ajouter un worker revient simplement à connecter un autre client avec le même filtre ; en retirer un, à le déconnecter. Donnez à chaque instance un client ID unique, comme expliqué dans les bonnes pratiques pour le client ID MQTT.
• Contrôle de flux. Combinez les abonnements partagés avec un petit Receive Maximum MQTT 5 sur chaque worker, afin qu'une instance lente ne reçoive pas une grosse rafale de messages non acquittés.
• Les wildcards sont acceptés. $share/group/factory/# répartit une branche entière. Les règles de correspondance sont les règles habituelles, expliquées dans les wildcards MQTT.
• Ne l'utilisez pas pour la diffusion. Si chaque consommateur doit voir chaque message, utilisez des abonnements classiques.

Vous découvrez MQTT ? Le tutoriel MQTT aborde dès le départ les abonnements, la QoS et les sessions.

Faire monter en charge vos consommateurs sur ConnectMQ

ConnectMQ est un broker MQTT managé compatible MQTT 3.1.1 et 5.0, avec des abonnements partagés disponibles à partir de l'offre Serverless (Paid), ainsi que la QoS 2, des messages de 1 Mo et jusqu'à 1 000 connexions. Les options Dedicated et BYOC sont disponibles lorsque vous avez besoin d'un cluster isolé. Inscrivez-vous et commencez avec l'offre Serverless gratuite sans carte bancaire, puis passez à l'offre supérieure lorsque votre backend a besoin de groupes de consommateurs. Connectez vos workers avec YOUR_USERNAME, YOUR_PASSWORD et YOUR_CLIENT_ID.

Questions fréquentes

Quelle est la syntaxe d'un abonnement partagé MQTT ?

Utilisez $share/<ShareName>/<TopicFilter>, par exemple $share/workers/sensors/+/temperature. Tous les clients qui utilisent le même nom de partage et le même filtre forment un groupe.

Les abonnements partagés fonctionnent-ils avec MQTT 3.1.1 ?

Ils sont définis dans MQTT 5. Certains brokers acceptent la syntaxe $share de la part de clients 3.1.1 en tant qu'extension, mais vérifiez-le dans la documentation de votre broker avant de vous y fier.

Les retained messages sont-ils livrés aux abonnements partagés ?

Non. Lorsqu'un abonnement partagé est créé, les retained messages ne lui sont pas envoyés. Lisez l'état retenu avec un abonnement classique distinct si un worker en a besoin.

Comment le broker choisit-il l'abonné qui reçoit un message ?

La spécification MQTT laisse ce choix au broker. Les sélections round-robin et aléatoire sont courantes : ne dépendez donc jamais du fait qu'un membre précis reçoive un message précis.