MQTT

MQTT はパブリッシュ/サブスクライブ方式でアプリケーションメッセージを伝送します：デバイスはブローカーのトピックにパブリッシュし、ブローカーはサブスクリプションフィルターに基づいてクライアントに配信します。伝送は信頼性が高く、順序が保証され、双方向の接続（一般的には TCP または WebSocket）に依存しており、信頼性の低い UDP 上で直接動作する同じパケットセットではありません。

温度センサーを例に挙げると、周期的な読み取り値はたまに1件欠落しても許容されますが、「バルブを開く」コマンドは実行されたかどうかの確認が必要です。MQTT QoS はプロトコルメッセージの配信を管理しますが、ビジネスコマンドにはコマンドID、有効期限、実行応答が必要です。これらは別々に設計すべきです。MQTT 5.0

Topic：名前、フィルター、権限は別々に処理する

Topic Name は大文字と小文字を区別し、/ で階層を区切ります。前後のスラッシュや空の階層にも意味があります。PUBLISH では + や # のワイルドカードを使用しません。サブスクリプションフィルターでは、+ は1つの階層にマッチし、# は末尾に位置し、0個以上の階層をカバーする必要があります。通常のトップレベルワイルドカードは、$ で始まるシステムトピックにはマッチしません。

命名はルーティングと認可の両方を考慮すべきです。例えば tenant/a/device/17/state はテナントとデバイスの範囲を明確に示せますが、クライアントがその文字列を宣言しただけでは、それがテナントaに属することを証明するものではありません。ブローカーの認証とACLは、パブリッシュおよびサブスクライブ可能な範囲を制限する必要があります。Mosquitto アクセス制御設定

QoS：各ホップが自身のパケット状態を管理する

パブリッシャーからブローカーへ、およびブローカーからサブスクライバーへは、それぞれ配信を交渉/実行します。送信側の QoS は、元の PUBLISH QoS、サブスクライブの許可上限、サーバーの能力によって制約されます。ソース側が QoS 2 を使用しているからといって、すべてのサブスクライバーが QoS 2 を取得できるとは限りません。

QoS 2 は、ビジネスデータベースと MQTT 状態が同じトランザクションを共有することを要求しません。アプリケーションが在庫を変更したが、消費結果の記録前にクラッシュした場合、復旧後の処理はクライアント、アプリケーションのコミット境界、およびメッセージの重複排除に依存します。デバイスコマンドの場合、イベントIDと実行結果を確実に関連付ける必要があります。Packet Identifier は交換が完了後に再利用可能であり、恒久的な重複排除キーとして使用することはできません。

MQTT 5 では、拒否理由付きの PUBACK/PUBREC も正常な成功として処理してはいけません。パブリッシャーがブローカーからの成功確認を受信しても、通常はそのホップの処理結果を示すだけで、すべてのサブスクライバーが受信したことを証明するものではなく、バルブが動作したことを証明するものでもありません。

再接続時に再送信するのは何か

MQTT 5 のプロトコル再送信は、主にセッション復旧中に確認されていない QoS 1/2 の PUBLISH および PUBREL に対して行われます。存続中の同一接続に対して、適当なタイムアウトによる定期的な再送信アルゴリズムを追加してはいけません。プロトコルレベルで PUBLISH を再送信する場合、DUP マークは重複する可能性があることを示しますが、DUP=0 の新しいパブリッシュでも同じビジネスイベントコンテンツを含む可能性があります。MQTT 5 §4.4

ネットワークが長期間到達不能であっても、セッションが期限切れになっても、管理者が状態を削除しても、プロトコルは最終的な配信を保証しません。QoS のセマンティクスは、状態の保持とエラー結果と組み合わせて考慮する必要があり、「重複なし・消失なし」を前提条件なしの運用約束として使用することはできません。

パケット長：2バイトの固定ヘッダーは最小限の場合のみ

MQTT パケットの最初のバイトにはタイプと固定フラグが含まれ、続く Remaining Length は1〜4バイトの可変長整数で残りの部分の長さを表します。127 を超えると、1バイトに収まらなくなります。トピック文字列の長さはバイトエンコーディングで表され、Packet Identifier は2バイトで、対応するメッセージタイプ/QoS の下でのみ出現します。

以下に、MQTT 5 の QoS 1 PUBLISH を構築します。トピック home/temp、Packet Identifier 7、属性なし、ペイロード 23.5。最初のバイトは 0x32、Remaining Length は 18、全体で 20 バイトです。TCP/TLS/IP は含まれていません。

import struct

def vbi(n):
    if not 0 <= n <= 268435455:
        raise ValueError('MQTT 可変長整数の範囲を超えています')
    result = bytearray()
    while True:
        digit = n % 128
        n //= 128
        result.append(digit | (0x80 if n else 0))
        if not n:
            return bytes(result)

assert vbi(127) == b'\x7f'
assert vbi(128) == b'\x80\x01'
assert vbi(16384) == b'\x80\x80\x01'
assert vbi(268435455) == b'\xff\xff\xff\x7f'
topic = b'home/temp'
variable_and_payload = struct.pack('!H', len(topic)) + topic + b'\x00\x07' + b'\x00' + b'23.5'
packet = b'\x32' + vbi(len(variable_and_payload)) + variable_and_payload
assert len(variable_and_payload) == 18 and len(packet) == 20
assert packet.hex() == '32120009686f6d652f74656d7000070032332e35'
print('MQTT 5 PUBLISH エンコーディングおよび Remaining Length 境界チェック通過')

Topic Alias の値は固定で2バイトです。MQTT 5 の属性識別子と属性長には追加のオーバーヘッドがあります。これは「1/2 バイトのエイリアス」ではありません。エイリアスマッピングは接続と方向ごとに管理され、再接続後にブローカーが古いエイリアスを覚えているとは限りません。最初のマッピング作成にはトピック名が必要で、その後で許可された範囲内で重複する名前を省略できます。

セッション、保持値、ウィルのライフサイクル

Clean Start と Session Expiry

MQTT 5 では、Clean Start によって接続時に新しいセッションを開始するかどうかを決定し、Session Expiry Interval によって切断後に状態を保持する期間を決定します。Clean Start=1 でも非ゼロの期限を設定でき、これは古いセッションをクリアして新しいセッションを保持することを意味します。Clean Start=0 は必ずしも復旧成功を意味せず、CONNACK の Session Present を確認する必要があります。

セッションを復旧するには、安定した Client ID を使用し、クライアント側の未完了の配信状態を復元する必要があります。同じ Client ID を同時に使用している2台のデバイスは互いの接続を追い出し、継続的な再接続を引き起こす可能性があります。また、接続ごとにランダムな ID を生成すると、復旧しようとしていたサブスクリプションやキューの関係を破壊してしまいます。

MQTT 3.1.1 の Clean Session と MQTT 5 のこれらのフィールドは直接混在させてはいけません。3.1.1 でもサーバーは特定の条件下で空文字列の Client ID を受け入れることが可能であり、「5.0 でなければサーバー割り当てができない」というわけではありません。MQTT 5 には、Assigned Client Identifier 属性を通じてクライアントに割り当て値を返す機能があります。MQTT 3.1.1

Retained：後発者へのスナップショット

RETAIN フラグ付きでパブリッシュされたメッセージは、そのトピックの保持値となり、後のサブスクライバーはサブスクライブオプションに基づいてそれを受け取ります。新しい値が古い値を置き換えるため、retained を使って過去の温度履歴をすべて再生することはできません。RETAIN 付きの空ペイロードは、そのトピックの保持メッセージをクリアするために使用されますが、現在のサブスクライバーは引き続きこのパブリッシュを受信する可能性があります。

Retained 状態は特定のサブスクライベントのセッションに属するものではなく、そのクライアントのセッションをクリアしても、すべてのトピックの保持値がクリアされるわけではありません。MQTT 5 の Message Expiry はメッセージの寿命を制限できるため、「サーバーが以前保持していた」からといって、新しいサブスクライバーが永遠に取得できるわけではありません。

Will：オフライン推論は即時の障害検出ではない

クライアントは CONNECT 中にウィルを登録します。正常な終了以外の条件がトリガーされると、ブローカーはルールに基づいてパブリッシュします。正常な切断では通常、ウィルはキャンセルされます。MQTT 5 では、ウィルを明示的にパブリッシュするための DISCONNECT 理由も定義されています。

Will Delay は一時的な再接続の時間を確保するために使用でき、セッションがタイムリーに復旧すれば誤ったオフライン通知を回避できます。セッションがそれより早く終了した場合、ウィルが設定された遅延を完全に待つとは限りません。オンラインステータスメッセージにはタイムスタンプまたは世代情報を付与し、サブスクライバーが古い通知を識別できるようにします。また、offline という retained メッセージを期限切れにならない事実として扱ってはいけません。

セッションが残っていても、コマンドや保持値が残っているとは限らないか？

20秒の時点ではセッションは存続していますが、10秒のコマンドはすでに期限切れです。サブスクライブを復元したからといって、それを実行してはいけません。Session Expiry を15秒に短縮すると、Will は30秒待つ必要がなくなります。retained は独自の期限に従って存続します。

Keep Alive、メッセージ期限切れ、スロークライアント

Keep Alive は、クライアントが MQTT 制御パケットを送信する間隔の制約であり、PUBLISH などの制御パケットがすでに存在すればアクティビティ要件を満たすため、K秒ごとに無条件で PINGREQ を送信する必要はありません。他のトラフィックがない場合に PINGREQ/PINGRESP を使用します。サーバーは、有効な Keep Alive の約1.5倍の期間内に制御パケットを受信しない場合、ルールに従って接続を閉じます。値が0の場合はこのメカニズムが無効になります。MQTT 5 サーバーは、Server Keep Alive を介して使用する値を指定することもできます。

Session Expiry、Message Expiry、Will Delay はそれぞれ異なるオブジェクトを管理します。セッションが1日間保存されていても、10秒間有効な制御コマンドが、翌日に再接続した後に引き続き実行できることを意味するものではありません。デバイスアプリケーションもコマンドの有効期限を確認する必要があり、転送キューの設定だけに依存してはいけません。

Receive Maximum は、対応する QoS 1/2 の並列配信数を制限しますが、すべての QoS 0 トラフィックを制限するものではなく、総バイトメモリ上限でもありません。Maximum Packet Size は単一パケットのサイズを制御します。ブローカーのキュー、クライアントのバッファ、ディスク永続化、パブリッシュ速度には、それぞれ独自の制約が必要です。

1秒あたり50件のバックログが発生し、各メッセージが実際に1 KiB を占用すると仮定すると、10分間のオフラインで約29.3 MiB になります。インデックスやプロトコル状態は含まれていません。セッションに非常に長い寿命を設定しても、キューを制限しない場合、自動的に安全なオフラインストレージにはなりません。

Shared Subscription とビジネス要求/応答

$share/workers/jobs/+ は、workers グループの共有サブスクリプションを表し、グループ内の1つのメンバーが対応する配信を受け取り、ワークロード分散に適しています。どのメンバーが選択されるかはサーバーのポリシーによって決定され、プロトコルはラウンドロビン、公平な処理時間、またはすべてのワーカー全体でのビジネス完了順序を保証しません。

MQTT 5 の Response Topic と Correlation Data は、アプリケーションが要求/応答を実現するのに役立ちますが、これらは自動的には同期 RPC のタイムアウト、認可、またはトランザクションを提供しません。要求側は、応答を受信する準備をし、要求にビジネス ID を割り当て、タイムアウト後に「操作は完了したが応答が消失した」という分岐を考慮する必要があります。

接続成功からビジネス有効性へ

ブローカーの認証は誰が接続できるかを解決し、topic ACL は何のパブリッシュまたはサブスクライブが可能かを解決し、payload スキーマとビジネス状態はメッセージが実行可能かどうかを解決します。Client ID はセッション識別子であり、信頼できるバインディングがない限り、デバイスIDとして直接使用すべきではありません。

Mosquitto などの実装における永続化とキューの設定は、再起動後の復旧とリソース制限に影響します。プロトコルの QoS は、確認応答が得られる前にディスクがあなたの災害復旧基準に従って永続化されたことを自動的に保証するものではありません。Mosquitto 設定

トラブルシューティング時には、Client ID、接続世代、トピック、ビジネスイベント ID、QoS、理由コード、Session Present を記録してください。応答のないパブリッシュについては、まず接続とフロー制御を確認します。パブリッシュ確認は得られたがビジネス結果がない場合は、サブスクライブマッチ、権限、期限切れ、消費コミットを継続して確認し、QoS を 1 から 2 に変更するだけでなく対処してください。