[MQTT ]Willについて

はじめに

Willについてみたあと、動作を確認しましょう。

Willについて

Willは遺言を意味します。Connect時に"遺言"となるPublishパケットを設定しておくことで、異常切断となった場合にサーバはそれがクライアントから送信されたのと同じ処理(=Publish)を行います。つまり、クライアントがDisconnectを送らずに接続が失われた状態になった場合にだけ実行されます。
例えば下記のような場合に実行されます。

• Ping/Pongタイムアウト
• TCP切断・TCP RST
• OS強制終了
• プロトコル違反
• サーバ側リソース枯渇による切断

逆に異常ではない正常切断とはクライアントがDisconnectを送って切断することです。

　

実験

この実験では異常切断を意図的に発生させるとBrokerがWillをPublish することを確認します。

• Connect時にWillを設定
• Disconnectを呼ばずにOSのネットワークソケットをCloseして異常切断を作りだす
• WillパケットがPublishされて受信

　[※1]ではpahoの内部ソケット参照が残っていると自動再接続などが走るため、明示的に無効化しています。

import paho.mqtt.client as mqtt
import threading
import time

BROKER = "localhost"
TOPIC = "test/topic"
WILL_TOPIC = "test/will"

# ======================
# Subscriber
# ======================
def on_message(client, userdata, msg):
    print(f"[受信] {msg.topic}: {msg.payload.decode()}")

def subscriber():
    sub = mqtt.Client(client_id="sub-client",
                      callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    sub.on_message = on_message
    sub.connect(BROKER, 1883, 60)

    sub.subscribe(TOPIC)
    sub.subscribe(WILL_TOPIC)

    sub.loop_forever()

# ======================
# Publisher（異常終了をシミュレート）
# ======================
def publisher_with_will():
    pub = mqtt.Client(client_id="pub-client",
                      callback_api_version=mqtt.CallbackAPIVersion.VERSION2)

    # Will 設定
    pub.will_set(WILL_TOPIC, "pub offline", qos=1, retain=False)
    pub.connect(BROKER, 1883, 10)

    for i in range(3):
        msg = f"Hello {i}"
        pub.publish(TOPIC, msg)
        print(f"[送信] {msg}")
        time.sleep(1)

    print("💀 Publisher の TCP接続だけ異常終了させます（プロセスは生きてる）")

    # --- ここが重要：TCP接続だけ kill する ---
    sock = pub.socket()
    if sock:
        sock.close()        # TCP FIN （Broker からは異常終了扱い）
    pub._sock = None        # [※1] paho の内部参照も破壊
    pub.loop_stop()         # ネットワークループを停止する

# ======================
# メイン処理
# ======================
if __name__ == "__main__":
    t = threading.Thread(target=subscriber, daemon=True)
    t.start()

    time.sleep(1)
    publisher_with_will()

    # Will を受信する時間を確保
    time.sleep(10)

[~/qiita/client]$uv run will.py 
[送信] Hello 0
[受信] test/topic: Hello 0
[送信] Hello 1
[受信] test/topic: Hello 1
[送信] Hello 2
[受信] test/topic: Hello 2
Publisher の TCP接続だけ異常終了させます
[受信] test/will: pub offline

TCP RSTが発生したあとにtest/willにPublishが発行されていることを確認できます。（実際には RST / FIN どちらになるかは環境や Broker 実装によりますが、どちらも “Disconnect なしで接続断” と判定されるかと思います）

まとめ

Willについて確認し、実際に動作確認を行いました