Tester les abonnements partagés MQTT et la répartition de charge

Un abonnement partagé ($share/<groupe>/<topic>) transforme plusieurs subscribers en pool de workers : chaque message du topic est livré à un seul membre du groupe au lieu de tous. Pour le tester, démarrez plusieurs workers dans le même groupe, publiez une séquence numérotée et vérifiez trois choses : chaque message est arrivé exactement une fois dans le groupe, la charge a été répartie, et rien n'a été perdu quand un worker a planté en plein flux.

Ce guide présente une vérification rapide en CLI, un test Python avec simulation de crash et les cas limites à vérifier. Les concepts sont expliqués dans les abonnements partagés MQTT. Ici, on se contente de les tester.

Ce que fait un broker correct

La stratégie de répartition dépend du broker. Mosquitto utilise le round robin. D'autres brokers proposent des stratégies aléatoires, sticky ou basées sur un hash, souvent configurables. Votre test doit vérifier une répartition équitable avec une tolérance, pas un motif exact, sauf si vous dépendez d'une stratégie précise.

Étape 0 : vérifier que le broker le prend en charge

Un broker MQTT 5 annonce la prise en charge dans le CONNACK avec la propriété Shared Subscription Available. Si la propriété est absente, la prise en charge est supposée. Si elle vaut 0, le broker rejettera ou ignorera les abonnements $share, et le SUBACK renvoie 0x9E Shared Subscriptions not supported. Dans tous les cas, la vraie preuve reste le test de répartition ci-dessous. L'outil de reason codes décode le SUBACK que vous obtenez.

Le nom du groupe est un seul niveau de topic : pas de /, + ni #. Le reste est un filtre de topic normal, donc les wildcards fonctionnent après le groupe : $share/workers/testmqtt/share/+/jobs. Vérifiez vos filtres avec le topic matcher avant d'accuser le broker.

Vérification rapide avec la CLI Mosquitto

Ouvrez deux terminaux avec des membres du même groupe et un troisième avec un subscriber normal :

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

mosquitto_sub -h broker.connectmq.com -V mqttv5 -q 1 \
  -t 'testmqtt/share/9b1e/jobs' -v

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

Les terminaux 1 et 2 se partagent les dix jobs (en round robin, les impairs d'un côté, les pairs de l'autre). Le terminal 3 les voit tous. Mettez toujours le topic entre guillemets simples : sans guillemets ou entre guillemets doubles, votre shell développerait $share comme une variable.

Test automatisé de répartition et de failover

Ce script paho-mqtt 2.x démarre trois workers dans un groupe, publie 100 messages QoS 1 numérotés et affiche le décompte par worker, les trous et les doublons. Avec --failover, il fait planter worker-0 à mi-parcours : le worker cesse de lire, ses derniers messages restent donc non acquittés, puis la connexion TCP est coupée sans DISCONNECT, comme lors d'un vrai crash.

import sys, time, uuid, collections
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/share/{RUN}/jobs"
GROUP = f"workers{RUN}"
WORKERS, MESSAGES, RATE = 3, 100, 10     # RATE in messages per second
FAILOVER = "--failover" in sys.argv          # crash worker-0 halfway through

got = collections.defaultdict(list)          # worker name -> message ids received

def make_worker(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{name}-{RUN}", protocol=mqtt.MQTTv5)

    def on_connect(cl, userdata, flags, rc, props):
        if getattr(props, "SharedSubscriptionAvailable", 1) == 0:
            print(f"{name}: CONNACK says shared subscriptions are not available")
        cl.subscribe(f"$share/{GROUP}/{TOPIC}", qos=1)

    def on_subscribe(cl, userdata, mid, reason_codes, props):
        print(f"{name}: SUBACK {reason_codes[0]}")

    c.on_connect, c.on_subscribe = on_connect, on_subscribe
    c.on_message = lambda cl, userdata, msg: got[name].append(int(msg.payload))
    c.connect(HOST, PORT)
    c.loop_start()
    return c

workers = {f"worker-{i}": make_worker(f"worker-{i}") for i in range(WORKERS)}
time.sleep(2)                                 # let all subscriptions settle

pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"share-pub-{RUN}", protocol=mqtt.MQTTv5)
pub.connect(HOST, PORT)
pub.loop_start()
for i in range(MESSAGES):
    if FAILOVER and i == MESSAGES // 2:
        crashed = workers.pop("worker-0")
        crashed.loop_stop()                   # stop reading: unacked messages stay in flight
        crashed.socket().close()              # drop TCP without DISCONNECT, like a crash
        print(f"crashed worker-0 after message {i - 1}")
    pub.publish(TOPIC, str(i), qos=1)
    time.sleep(1 / RATE)
time.sleep(5)

all_ids = [m for ids in got.values() for m in ids]
missing = sorted(set(range(MESSAGES)) - set(all_ids))
dupes = len(all_ids) - len(set(all_ids))
for name in sorted(got):
    print(f"{name}: {len(got[name]):4d} messages, in order: {got[name] == sorted(got[name])}")
print(f"delivered {len(set(all_ids))}/{MESSAGES}, missing {len(missing)}, duplicates {dupes}")
if missing:
    print("first missing ids:", missing[:10])
for c in workers.values():
    c.disconnect(); c.loop_stop()
pub.disconnect(); pub.loop_stop()

Une exécution contre un broker en round robin ressemble à ceci :

worker-0: SUBACK Granted QoS 1
worker-1: SUBACK Granted QoS 1
worker-2: SUBACK Granted QoS 1
crashed worker-0 after message 49
worker-0:   17 messages, in order: True
worker-1:   42 messages, in order: True
worker-2:   41 messages, in order: True
delivered 100/100, missing 0, duplicates 0

Comment le lire :

• Sans --failover, les décomptes doivent être proches (33/33/34 en round robin). Une répartition très déséquilibrée indique une stratégie sticky ou par hash, ou un worker lent que le broker saute.
• Avec --failover, missing 0 signifie que le broker a redirigé le travail du membre planté. Des IDs manquants juste autour du crash signifient que des messages en cours ont été perdus avec la session. Certains brokers attendent d'abord que le client se reconnecte, testez donc avec vos vrais paramètres de session.
• Les doublons sont légaux en QoS 1 : un message peut être livré au worker planté et renvoyé à un autre. Les workers d'un groupe partagé doivent être idempotents. Voir les messages MQTT dupliqués.
• Des messages manquants sans aucun crash révèlent en général une limite de débit ou de file côté broker. Baissez RATE et relancez. Les endpoints publics partagés sont particulièrement susceptibles de brider.

Cas limites à vérifier

1. Ordre. Chaque membre voit ses propres messages dans l'ordre, mais il n'y a pas d'ordre global dans le groupe. Le message 41 peut être traité avant le 40 sur un autre worker. Si l'ordre compte par appareil, utilisez une stratégie de hash par topic quand elle existe ou partitionnez les topics. Le test d'ordre des messages montre comment le mesurer.
2. Messages retenus. Publiez un message retenu, puis démarrez un subscriber partagé : il ne doit rien recevoir. Les workers qui ont besoin d'un état initial doivent l'obtenir autrement.
3. QoS mixte. Le QoS livré est le minimum entre le QoS de publication et le QoS accordé à chaque membre. Un membre abonné en QoS 0 perd des messages en cas de crash, même si les autres utilisent le QoS 1.
4. Options d'abonnement. No Local sur un abonnement partagé est une erreur de protocole. Les autres options sont expliquées dans les options d'abonnement MQTT.
5. Faire grandir le groupe. Ajoutez un quatrième worker en cours de route : il doit recevoir sa part immédiatement. Retirez-en un : les autres absorbent la charge.
6. Membres persistants. Avec un Session Expiry supérieur à 0, un membre déconnecté garde sa place dans le groupe, et certains brokers continuent de lui envoyer des messages qui s'accumulent pendant qu'il est hors ligne. Décidez si vos workers doivent vraiment être persistants.

Tester le gain de débit

Les abonnements partagés existent pour faire monter en charge les consommateurs. Mesurez que le gain est réel : lancez le script avec 1, 2 et 4 workers qui attendent un temps de traitement fixe par message, et comparez le temps total pour vider une rafale. Le débit doit croître de façon presque linéaire jusqu'à ce que le broker ou le réseau devienne la limite. Un test de fan-in en est le complément naturel : de nombreux publishers alimentant un groupe.

Observer la répartition en direct

Ouvrez le client MQTT en ligne dans deux onglets, abonnez-les au même filtre $share/... et publiez depuis un troisième onglet. Vous voyez la répartition en temps réel, une bonne vérification avant de lancer le script.

Publier des jobs de test depuis le navigateur

Un test d'abonnements partagés prend quelques minutes et répond aux questions qui comptent en production : chaque job est-il traité, la charge est-elle vraiment répartie, et que devient le travail en cours quand un worker meurt ? Gardez l'exécution de failover dans votre suite de CI et relancez-la à chaque changement de broker ou de paramètres de session.

Questions fréquentes