mqtt Advent Calendar 2025

[MQTT] Message Expiry

Last updated at 2025-12-22Posted at 2025-12-22

Message Expiryとは

PublishのVariable Headerに格納するメタデータで、そのPublishパケット(Application Message)の有効期限を指定します。指定しない場合は無期限となります。
　Publishパケットが保存されるのは下記の2つの場合です

Retainフラグを設定した場合
QoS1/QoS2でオフラインセッション中に滞留している場合(QoS0はOPTIONAL)

例えば、そもそも下記の場合は保存されません。有効期限は関係ありません。

QoS0 かつ Retainを設定していないPublish

確認

今回はRetainを使って保持します。有効期限の長いPublishと有効期限の短いPublishを発行し、Subscribe開始タイミングを調整することで有効期限の長いPublishのみがRetainによって受信されることを確認します。

サーバ側は1日目のものを使います
クライアント

import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes
import threading
import time

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

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

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

    print("Subscriber: 5秒待ってから subscribe")
    time.sleep(5)

    sub.subscribe(TOPIC)
    print("Subscriber: subscribe 完了")
    sub.loop_forever()

# ======================
# Publisher
# ======================
def publish(pub, payload, expiry):
    props = Properties(PacketTypes.PUBLISH)
    props.MessageExpiryInterval = expiry

    pub.publish(
        TOPIC,
        payload=payload,
        qos=0,
        retain=True,  # ← 重要
        properties=props,
    )
    print(f"[送信] {payload} (retain, expiry={expiry}秒)")

def publisher():
    pub = mqtt.Client(
        client_id="pub-client",
        protocol=mqtt.MQTTv5,
        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
    )
    pub.connect(BROKER, 1883)

    # すぐ期限切れになる Retain メッセージ
    publish(pub, "Hello short", expiry=2)

    time.sleep(1)

    # Subscribe 時点でも生きている Retain メッセージ
    publish(pub, "Hello long", expiry=10)

    pub.disconnect()

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

    time.sleep(1)
    publisher()

    time.sleep(10)

著作権情報

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up