MQTT Streams クイックスタート
このページでは、EMQX 6.1 の MQTT Streams 機能の使い方を説明します。MQTTX を使ってクライアントをシミュレートし、EMQX ダッシュボードからストリームを作成・管理し、メッセージの保存、再生、圧縮の動作を体験します。
目的
このクイックスタートでは、EMQX MQTT Streams が以下を実現できることを示します。
- サブスクライバーの有無に関わらずメッセージを永続化する
- タイムスタンプに基づく再生をサポートする
- 状態指向のメッセージングに適した Last-Value セマンティクスを有効にする
$stream/プレフィックスのサブスクライブによりストリームを自動作成する
前提条件
開始前に以下を準備してください。
- EMQX 6.1 以上が稼働していること
- MQTTX（または MQTT 5.0 対応のクライアント）
- EMQX ダッシュボードへのアクセス（デフォルト：
http://localhost:18083）
MQTT Streams の基本機能を試す（レギュラーストリーム）
このセクションでは、MQTT Streams がメッセージを保存し、コンシューマーが過去のデータを再生できることを示します。
前提条件
開始前に、MQTT Streams 機能が有効であり、自動作成の設定が本例に影響しないことを確認してください。
左メニューの Streams に移動します。
Message Stream が無効の場合は、Settings をクリックします。Management -> MQTT Settings -> Streams ページにリダイレクトされます。
Enable Streams スイッチをオンにします。
自動作成設定を確認し、Regular Stream が使用されるようにします。
Enable Auto Create Streams が無効、または
Auto Create Stream Type が Regular Stream に設定されていること。
これにより、ストリームが Last-Value Stream として自動作成されず、キーごとに最新のメッセージのみを保持する動作を防ぎます。
変更した場合は、Save Changes をクリックして適用します。
ステップ 1: 名前付きストリームを作成する
左メニューの Streams に移動します。
ページ内の Create Stream をクリックするか、右上の Create をクリックします。
Create Stream ダイアログで以下を設定します。
- Name:
my_stream - Topic Filter:
demo/stream - Last-Value Semantics: 無効
- Stream Key Expression:
message.from
その他のオプションはデフォルトのままにします。
- Name:
Create をクリックします。
ステップ 2: メッセージをパブリッシュする
MQTTX CLI を使ってパブリッシャークライアントをシミュレートします。
MQTTX CLI がインストールされていることを確認します。詳細は インストール を参照してください。
EMQX に接続します。
bashmqttx conn -h 'localhost' -p 1883QoS 1 でトピック
demo/streamに複数のメッセージをパブリッシュします。例:
bashmqttx pub -t 'demo/stream' -h 'localhost' -p 1883 -q 1 -m '{"value": 1}' mqttx pub -t 'demo/stream' -h 'localhost' -p 1883 -q 1 -m '{"value": 2}' mqttx pub -t 'demo/stream' -h 'localhost' -p 1883 -q 1 -m '{"value": 3}'期待される出力:
bash✔ Connected ✔ Message published
このストリームはレギュラーストリームなので、すべてのパブリッシュされたメッセージは保持ポリシーに従ってストリームに保存されます。
ステップ 3: ストリームからすべてのメッセージを再生する
MQTTX CLI を使い、MQTT 5 のサブスクリプションユーザープロパティ stream-offset を設定してストリームの先頭からメッセージを再生します。
mqttx sub -t \$stream/my_stream -q 1 -h localhost -up "stream-offset: 0"期待される動作:
これまでにパブリッシュされたすべてのメッセージがパブリッシュ順に受信されます。
topic: demo/stream, qos: 0, size: 10B, userProperties: [
{ key: 'key', value: 'mqttx_28c50267' },
{ key: 'ts', value: '1772161077594532' }
]
{"value": 1}
topic: demo/stream, qos: 0, size: 10B, userProperties: [
{ key: 'key', value: 'mqttx_1989d120' },
{ key: 'ts', value: '1772161084921509' }
]
{"value": 2}
topic: demo/stream, qos: 0, size: 10B, userProperties: [
{ key: 'key', value: 'mqttx_085ea00d' },
{ key: 'ts', value: '1772161094020511' }
]
{"value": 3}これにより以下が確認できます。
- ストリームはレギュラーストリームである
stream-offsetサブスクリプションプロパティが正しく機能している- メッセージは圧縮や上書きされていない
異なる位置からメッセージを再生する
MQTT Streams では、サブスクライバーが stream-offset の値を指定して再生開始位置を制御できます。
ここでは、特定の時点以降にパブリッシュされたメッセージのみを再生する例を示します。
ステップ 1: 現在のタイムスタンプを取得する
新しいメッセージをパブリッシュする前に、現在の Unix タイムスタンプ（マイクロ秒単位）を記録します。
ミリ秒単位の現在時刻は以下で取得可能です。
Linux / macOS:
bashdate +%s000JavaScript:
javascriptDate.now()
例:
1772162409000この値に 1000 を掛けてマイクロ秒単位に変換し、保存します。これが再生開始位置になります。
ステップ 2: 新しいメッセージをパブリッシュする
ストリームにさらにメッセージをパブリッシュします。
mqttx pub -t 'demo/stream' -h 'localhost' -p 1883 -q 1 -m '{"value": 4}'
mqttx pub -t 'demo/stream' -h 'localhost' -p 1883 -q 1 -m '{"value": 5}'ステップ 3: 記録したタイムスタンプから再生する
保存したタイムスタンプを stream-offset として指定してストリームにサブスクライブします。
mqttx sub -t \$stream/my_stream -q 1 -h localhost -up "stream-offset: 1772162409000000"期待される動作:
この時刻以降にパブリッシュされたメッセージのみが配信されます。
topic: demo/stream, qos: 1, size: 12B, userProperties: [
{ key: 'key', value: 'mqttx_a5508c54' },
{ key: 'ts', value: '1772163340159513' }
]
{"value": 4}
topic: demo/stream, qos: 1, size: 12B, userProperties: [
{ key: 'key', value: 'mqttx_e0848366' },
{ key: 'ts', value: '1772163350666523' }
]
{"value": 5}異なるコンシューマーは同じストリームを別々の位置から再生できます。
- あるコンシューマーは先頭（
earliest）から再生 - 別のコンシューマーは特定のタイムスタンプから再生
- さらに別のコンシューマーは新しいメッセージのみ（
latest）を受信
各コンシューマーの再生位置は独立しており、他のサブスクライバーに影響を与えません。
これにより、MQTT Streams のコンシューマー制御型再生が実証されます。
Last-Value セマンティクスを試す
このセクションでは、Last-Value MQTT Streams がキーごとに最新のメッセージのみを保持し、状態を表現するのに適していることを示します。
ステップ 1: 既存のストリームを削除する
- ダッシュボードの Streams に移動します。
my_streamを探します。- Delete をクリックし、確認します。
ステップ 2: Last-Value メッセージストリームを作成する
Streams ページで Create をクリックします。
以下を設定します。
- Name:
device_stream - Topic Filter:
device/state - Data Retention Period:
7日 - Last-Value Semantics: 有効
- Stream Key Expression:
message.from
- Name:
Create をクリックします。
キー式が message.from のため、同じキーのストリームでは最新のメッセージのみが保持されます。
ステップ 3: 状態更新をパブリッシュする
同じクライアント ID -i device-1 からメッセージをパブリッシュします。
mqttx pub -t 'device/state' -h 'localhost' -p 1883 -q 1 -i device-1 -m '{"status": "online"}'
mqttx pub -t 'device/state' -h 'localhost' -p 1883 -q 1 -i device-1 -m '{"status": "offline"}'ストリームキー式がメッセージメタデータのクライアント ID を抽出してキーとしているため、両メッセージは同じキーを持ち、2つ目のメッセージが1つ目を上書きします。
ステップ 4: ストリームにサブスクライブする
ストリームにサブスクライブし、先頭から再生します。
mqttx sub -t '$stream/device_stream' -h 'localhost' -p 1883 -q 1 -up "stream-offset: 0"期待される動作:
最新のメッセージのみが配信されます。
topic: device/state, qos: 1, size: 21B, userProperties: [
{ key: 'key', value: 'device-1' },
{ key: 'ts', value: '1772173666097076' }
]
{"status": "offline"}これにより、MQTT Streams が Last-Value セマンティクスを用いた状態指向メッセージングをサポートしていることが示されます。
ストリームの自動作成
EMQX の MQTT Streams は、クライアントが $stream/ プレフィックス付きトピックにサブスクライブするとストリームを自動作成できます。これにより、ダッシュボードで手動作成せずに動的にストリームをプロビジョニング可能です。
このセクションでは、自動作成を有効にして動作を確認します。
ダッシュボードの Management -> MQTT Settings -> Streams に移動します。
Enable Auto Create Streams がオンになっていることを確認します。
ストリームタイプを選択します。
- Regular Stream
- Last-Value Stream
自動作成で有効にできるのは一度に1種類のみです。
その他のオプションはデフォルトのままにします。
Save Changes をクリックします。
以下のコマンドでサブスクライブし、自動作成をトリガーします。
bashmqttx sub -h localhost -p 1883 -q 1 -t '$stream/auto_stream/demo/auto' -up "stream-offset: earliest"手動作成ストリームと異なり、自動作成ストリームはサブスクライブ時にトピックフィルター（この例では
demo/auto）が必要です。ストリームが存在しない場合、EMQX は以下を実行します。
auto_streamという名前の新しいストリームを作成- トピックフィルターを
demo/autoに設定 - 設定された自動作成タイプ（レギュラーまたは Last-Value）を適用
ダッシュボードの Streams ページで自動作成が確認できます。ストリーム一覧に
auto_streamが表示されます。