MQTT 設定

MQTT は、モノのインターネット（IoT）デバイスを接続するための軽量なパブリッシュ／サブスクライブ型メッセージングプロトコルです。EMQX は MQTT 3.1、3.1.1、および 5.0 をサポートしています。

このページでは、EMQX における MQTT プロトコルの動作設定について、パケット検証と制限、サブスクリプション、遅延パブリッシュ、キープアライブ処理、セッションなどを説明します。

基本的な MQTT 設定

このセクションでは、パケットサイズ、クライアントID長、トピックレベル数、QoS（サービス品質）、トピックエイリアス、リテインの設定など、MQTT プロトコルの動作を決定する設定項目を紹介します。

TIP

対応する設定項目は EMQX ダッシュボードの Management -> MQTT Settings -> General でも確認できます。ダッシュボードで設定した場合、その設定は設定ファイル内の同じ項目より優先されます。
設定ファイルから MQTT を設定する場合は、emqx.conf ではなく base.hocon の使用を推奨します。
emqx.conf に設定した場合、ダッシュボードでの変更は一時的なものとなり、EMQX の再起動時に失われるためです。

設定例:

bash

mqtt {
  max_packet_size = 1MB
  max_clientid_len = 65535
  max_topic_levels = 128
  max_qos_allowed = 2
  max_topic_alias = 65535
  retain_available = true
  strict_mode = true
}

各項目の説明は以下の通りです。

設定項目	ダッシュボード表示名	説明	デフォルト値	設定可能値
`max_packet_size`	Max Packet Size	MQTT パケットは MQTT クライアントと EMQX 間でメッセージを送受信するために使用されます。許可される最大 MQTT パケットサイズを設定します。	`1MB`
`max_clientid_len`	Max Client ID Length	MQTT クライアントIDの最大長を設定します。過度に長いクライアントIDの使用を防止し、問題発生を抑制します。	`65535`	`23` - `65535`
`max_topic_levels`	Max Topic Levels	MQTT トピックはメッセージの分類・整理に使われます。トピックの最大レベル数を設定します。	`128`	`1` - `35`
`max_qos_allowed`	Max QoS	QoS（サービス品質）レベルはメッセージの信頼性や配信保証のレベルを決定します。許可される最大 QoS レベルを設定します。
`max_topic_alias`	Max Topic Alias	トピックエイリアスは、完全なトピック名の代わりに短いエイリアスを使うことで MQTT パケットサイズを削減する仕組みです。 MQTT セッションで使用可能な最大トピックエイリアス数を設定します。	`65535`	`1` - `65535`
`retain_available`	Retain Available	リテインメッセージは、トピックに最後にパブリッシュされたメッセージを保存し、新規サブスクライバーが最新メッセージを受信できるようにします。リテイン機能を有効にするかどうかを設定します。	`true`	`true`, `false`
`strict_mode`	Strict Mode	受信した MQTT パケットに対して追加のプロトコル準拠チェックを適用するかどうかを設定します。チェックに失敗したパケットはクライアント接続を切断します。	`true`	`true`, `false`

厳格な MQTT パケット検証

EMQX 6.3.0 以降、strict_mode = true がデフォルトで有効になっており、以下のような不正な MQTT パケットを EMQX が拒否します。

不正な MQTT 固定ヘッダーフラグの組み合わせ。
MQTT 3.1.1 の CONNECT パケットにおいて、Username Flag なしに Password Flag が設定されている場合。
クライアントID、トピック名、ユーザー名、パスワード、Will トピック、MQTT 5.0 の文字列プロパティなどに含まれる不正な UTF-8 文字列（ヌル文字や禁止されている制御文字を含む）。
MQTT プロトコルで非ゼロ値が要求されるパケット識別子がゼロの場合。

不正なパケットを検出すると、EMQX はクライアント接続を切断し、info レベルのログに frame_parse_error と具体的な理由を記録します。MQTT 5.0 クライアントには可能な場合、理由コード 0x81（Malformed Packet）を含む CONNACK または DISCONNECT パケットを送信します。MQTT 3.1 および 3.1.1 クライアントは理由コードなしで切断されます。

既存のクライアントがこれらの MQTT プロトコル要件に準拠していない場合、一時的に厳格モードのプロトコル準拠チェックを無効にできます。

bash

mqtt.strict_mode = false

特定のレガシークライアントのみで厳格モードを無効にするには、ゾーンを設定し専用リスナーに関連付けます。

bash

zones.legacy_clients {
  mqtt.strict_mode = false
}

listeners.tcp.legacy {
  bind = "0.0.0.0:1884"
  zone = legacy_clients
}

他のリスナー経由で接続するクライアントは引き続き厳格な検証を使用します。ゾーンの詳細は Zone Override を参照してください。

サブスクリプション設定

EMQX におけるサブスクリプションとは、クライアントが EMQX のトピックにサブスクライブするプロセスを指します。クライアントがトピックにサブスクライブすると、そのトピックにパブリッシュされたメッセージを受信したいことを示します。

このセクションでは、共有サブスクリプション、ワイルドカードサブスクリプション、排他サブスクリプションの設定方法を紹介します。

TIP

設定例:

bash

mqtt {
	wildcard_subscription = true
  exclusive_subscription = false
  shared_subscription = true
  shared_subscription_strategy  =  round_robin
}

各項目の説明は以下の通りです。

設定項目	ダッシュボード表示名	説明	デフォルト値	設定可能値
`wildcard_subscription`	Wildcard Subscription Available	ワイルドカードサブスクリプションは、`+` や `#` といったワイルドカードを使い、複数トピックに一括サブスクライブ可能にします。ワイルドカードサブスクリプションを有効にするかどうかを設定します。	`true`	`true`, `false`
`exclusive_subscription`	Exclusive Subscription	排他サブスクリプションは、一度に1つの MQTT クライアントのみがトピックにサブスクライブ可能にします。排他サブスクリプションを有効にするかどうかを設定します。	`true`	`true`, `false`
`shared_subscription`	Shared Subscription Available	共有サブスクリプションは複数の MQTT クライアントがトピックのサブスクリプションを共有できます。共有サブスクリプションを有効にするかどうかを設定します。	`true`	`true`, `false`
`shared_subscription_strategy`		共有サブスクリプションのメッセージ配信戦略を定義します。 `shared_subscription` が `true` の場合にのみ必要です。	`round_robin`	- `random`（ランダムにサブスクライバーへ配信） - `round_robin`（ラウンドロビン方式で順番に配信） - `sticky`（最後に選択されたサブスクライバーへ常に配信、切断時まで維持） - `hash`（`clientIds` のハッシュに基づいてサブスクライバーを選択）

遅延パブリッシュ設定

遅延パブリッシュ機能は、クライアントがメッセージのパブリッシュを指定時間遅延させることを可能にします。この機能は、特定の時間にメッセージをパブリッシュしたい場合や、特定条件が満たされたときにメッセージを送信したい場合に有用です。

このセクションでは、遅延パブリッシュの有効化方法と許可される最大遅延メッセージ数の設定方法を紹介します。

設定例:

bash

delay {
  delayed_publish_enabled = true
  max_delayed_messages = 0
}

各項目の説明は以下の通りです。

delayed_publish_enabled は遅延パブリッシュ機能を有効にするかどうかを設定します。デフォルト値は true、設定可能値は true または false です。
max_delayed_messages は許可される最大遅延メッセージ数を設定します。デフォルト値は 0 です。

キープアライブ設定

キープアライブは 2 バイトの整数で、秒単位の時間間隔を表します。これは、MQTT クライアントと EMQX 間の接続がデータ送受信がなくてもアクティブであることを保証する仕組みです。MQTT クライアントが EMQX に接続する際、CONNECT パケットのヘッダーに非ゼロのキープアライブ値を設定すると、双方の間でキープアライブ機能が有効になります。詳細は MQTT キープアライブパラメータとは？を参照してください。

MQTT 5.0 プロトコルによると、キープアライブが有効なクライアントに対し、サーバーはキープアライブ時間の1.5倍の間クライアントから MQTT コントロールパケットを受信しなければ、ネットワーク接続を切断しなければなりません。
そのため、EMQX はクライアントのキープアライブタイムアウト状態を定期的にチェックするための設定 keepalive_multiplier を導入しています。デフォルト値は 1.5 です。

bash

keepalive_multiplier = 1.5

タイムアウト計算式は以下の通りです。

Keep Alive \times keepalive_multiplier

動的キープアライブ調整

車載ネットワーク（T-Box）やモバイル IoT などのシナリオでは、MQTT クライアントは「アクティブ状態」（頻繁な通信）と「スリープ状態」（低消費電力の待機）を切り替える必要があります。単一の固定キープアライブ値では両方のニーズを満たせません。

短いキープアライブはアクティブ時の切断検出を迅速にしますが、駐車中や待機時にはハートビート通信が過剰になりバッテリーを消耗します。
長いキープアライブはスリープ時の通信量を減らしますが、アクティブ時の切断検出が遅れます。

EMQX は $SETOPTS/ システムトピック群を通じて、クライアント単位での動的なキープアライブ調整をサポートします。クライアントはこれらのトピックにパブリッシュして自身のブローカー側キープアライブ許容値を更新でき、また権限を持つバックエンドサービスは複数クライアントを一括更新できます。MQTT 接続の切断や再交渉は不要です。調整はアクティブなセッションのメモリ上でのみ適用され、永続化されません。

リスナーのマウントポイントとの非互換性

動的キープアライブ調整は、マウントポイントが設定されたリスナー経由で接続するクライアントには機能しません。EMQX はマウントポイントを $SETOPTS/ プレフィックスの前に適用するため、更新はマウントされたリテラルトピックへの通常メッセージとしてルーティングされ、クライアントにエラーは通知されません。

単一クライアント更新：`$SETOPTS/mqtt/keepalive`

クライアントはこのトピックにパブリッシュして自身のブローカー側キープアライブタイムアウトを更新します。EMQX はパブリッシュ元のセッションからクライアントIDを自動的に取得します。

ペイロード: 非負整数（秒単位）の文字列。

text

有効範囲: 0～65535 秒。0 はそのセッションのキープアライブチェックを無効化します。65535 を超える値は 65535 にクランプされます。クライアントのゾーンで mqtt.server_keepalive が設定されている場合、実効値は両者の最小値です。

利用例: 車両が駐車状態に入る際、T-Box クライアントが $SETOPTS/mqtt/keepalive に 300 をパブリッシュします。EMQX はブローカー側のキープアライブ許容値を 300 秒（デフォルトの 1.5× 倍数で実効アイドルタイムアウトは 450 秒）に延長し、リモートコマンド配信のために MQTT 接続を維持します。なおこれはブローカー側のタイムアウト調整のみで、クライアントの実際の PINGREQ 間隔は変わりません。ハートビート通信量を減らすにはクライアント側もキープアライブ間隔を延長する必要があります。

一括更新：`$SETOPTS/mqtt/keepalive-bulk`

バックエンドサービスはこのトピックにパブリッシュして複数クライアントのキープアライブを一括更新できます。

ペイロード: JSON 配列。各要素は以下を含みます。

フィールド	型	必須	説明
`clientid`	文字列	必須	対象 MQTT クライアント識別子
`keepalive`	整数	必須	新しいキープアライブ間隔（秒、0～65535）

json

[
  { "clientid": "tbox-001", "keepalive": 300 },
  { "clientid": "tbox-002", "keepalive": 60 }
]

一括更新は非同期で処理され、クラスター対応です。EMQX は各対象クライアントをホストするノードを特定し、ノード間 RPC で更新を適用します。内部キューに 10 件以上のリクエストが溜まると追加リクエストは破棄され、警告ログが記録されます。

アクセス制御

2 つのトピックは細かい ACL をサポートするために分離されています。

認証済みクライアントに $SETOPTS/mqtt/keepalive へのパブリッシュを許可し、各デバイスが自身のキープアライブを調整可能にします。
$SETOPTS/mqtt/keepalive-bulk は信頼されたバックエンドサービスのみに制限します。

TIP

信頼できないクライアントに $SETOPTS/mqtt/keepalive へのパブリッシュ権限を与えないでください。キープアライブを 0 に設定するとそのセッションのキープアライブチェックが完全に無効化され、過度に大きな値は切断検出を遅延させてブローカーリソースを消費し続ける原因となります。

これらのトピックにパブリッシュされたメッセージは EMQX によってインターセプトされ消費され、サブスクライバーには配信されません。

セッション設定

MQTT におけるセッションとは、クライアントとブローカー間の接続状態を指します。EMQX では、クライアントが接続するとセッションが確立され、トピックのサブスクライブやメッセージの受信、EMQX へのメッセージパブリッシュが可能になります。

このセクションではセッションの設定方法を紹介します。

設定例:

bash

mqtt {
    max_subscriptions = infinity
    upgrade_qos = false
    max_inflight = 32
    retry_interval = 30s
    max_awaiting_rel = 100
    await_rel_timeout = 300s
    session_expiry_interval = 2h
    max_session_expiry_interval = infinity
    max_mqueue_len = 1000
    mqueue_priorities = disabled
    mqueue_default_priority = lowest
    mqueue_store_qos0 = true
    force_shutdown {
      max_mailbox_size = 1000
      max_heap_size = 32MB
    }
    force_gc {
      count  =  16000
      bytes  =  16MB
    }
  }

各項目の説明は以下の通りです。

設定項目	ダッシュボード表示名	説明	デフォルト値	設定可能値
`max_subscriptions`	Max Subscriptions	クライアントが保持可能な最大サブスクリプション数を設定します。	`infinity`	`1` - `infinity`
`upgrade_qos`	Upgrade QoS	メッセージパブリッシュ後に QoS（サービス品質）レベルをアップグレード可能にするか設定します。	`false`（無効）	`true`, `false`
`max_inflight`	Max Inflight	QoS 1 および QoS 2 メッセージの最大同時送信数（送信済みで未アック）を設定します。	`32`	`1` - `65535`
`retry_interval`	Retry Interval	QoS 1 または QoS 2 メッセージの再送間隔を設定します。	`30s` 単位: 秒	--
`max_awaiting_rel`	Max Awaiting PUBREL	各セッションで `PUBREL` 受信またはタイムアウトまで保留可能な QoS 2 メッセージ数を設定します。この上限に達すると、新規 QoS 2 `PUBLISH` はエラーコード `147(0x93)` で拒否されます。 MQTT の `PUBREL` は QoS 2 メッセージフローでの制御パケットです。	`100`	`1` - `infinity`
`await_rel_timeout`	Max Awaiting PUBREL TIMEOUT	QoS 2 メッセージの `PUBREL` 受信待ち時間を設定します。この時間を超えると EMQX はパケットIDを解放し、警告ログを出力します。注：`PUBREL` の有無に関わらず、EMQX は QoS 2 メッセージの転送を継続します。	`300s` 単位: 秒	--
`session_expiry_interval`	Session Expiry Interval	クライアント切断後に EMQX がセッションを保持する期間を設定します。 MQTT 3.1 および 3.1.1 クライアントの `Clean Session = false` 時に適用されます。MQTT 5.0 クライアントは CONNECT プロパティ `Session-Expiry-Interval` で独自に設定します（`max_session_expiry_interval` を参照）。デフォルトのインメモリセッションストアでは、切断されたセッションはこの期間メモリに保持されます。以下の警告も参照してください。	`2h`	--
`max_session_expiry_interval`	Max Session Expiry Interval	MQTT 5.0 クライアントが CONNECT および DISCONNECT パケットで要求可能なセッション有効期限の上限を設定します。接続時により長い値を要求した場合、EMQX はこの値にクランプし、CONNACK の `Session-Expiry-Interval` プロパティで返します。DISCONNECT パケットの長い値も同様にクランプされます。MQTT 3.1 および 3.1.1 クライアントには影響しません。 EMQX 6.3.0 以降で利用可能です。	`infinity`（無制限）	期間または `infinity`
`max_mqueue_len`	Max Message Queue Length	インメモリセッションのメッセージキュー長制限を設定します。クライアントがオフラインでセッションが残存している場合、インフライトウィンドウが満杯、または送信キューが混雑している場合にメッセージがキューに入ります。キューが制限に達すると、EMQX は優先度ごとに最も古い QoS 0 メッセージを優先的に削除します。	`1000`	`0` - `infinity`
`mqueue_priorities`	Topic Priorities	トピック優先度を設定します。ここでの設定は `mqueue_default_priority` の設定を上書きします。	`disabled` セッションは `mqueue_default_priority` の優先度を使用します。	`disabled` または `1` - `255`
`mqueue_default_priority`	Default Topic Priorities	デフォルトのトピック優先度を設定します。	`lowest`	`highest`， `lowest`
`mqueue_store_qos0`	Store QoS 0 Message	クライアントがオフライン、送信キューが混雑、またはインフライトウィンドウが満杯の間、EMQX がインメモリセッションのメッセージキューに QoS 0 メッセージを保存するかどうかを設定します。無効にすると、オフラインまたは送信キュー混雑時に到着した QoS 0 メッセージは破棄されます。インフライトウィンドウが満杯の場合は即時配信を継続します。	`true`	`true`, `false`
`force_shutdown`	Enable Force Shutdown	フォースシャットダウン機能を有効にするか設定します。メールボックスキュー長（`max_mailbox_size`）またはヒープサイズ（`max_heap_size`）が指定値に達すると、クライアント接続処理を強制終了します。	`true`	`true`, `false`
`force_shutdown.max_mailbox_size`	Max Mailbox Size	フォースシャットダウンをトリガーする最大メールボックスキュー長を設定します。	`1000`	`1` - `infinity`
`force_shutdown.max_heap_size`	Max Heap Size	フォースシャットダウンをトリガーする最大ヒープサイズを設定します。	`32MB`	--
`force_gc`	--	指定されたメッセージ数（`count`）または受信バイト数（`bytes`）に達した場合に強制ガベージコレクションを有効にするか設定します。	`true`	`true`, `false`
`force_gc.count`	--	強制ガベージコレクションをトリガーする受信メッセージ数を設定します。	`16000`	`0` - `infinity`
`force_gc.bytes`	--	強制ガベージコレクションをトリガーする受信バイト数を設定します。	`16MB` 単位: `MB`	--

切断されたセッションのメモリコスト

デフォルトのインメモリセッションストアでは、セッション有効期限がゼロより大きい場合、クライアント切断時にセッションは削除されません。EMQX はセッション、サブスクリプション、メッセージキューをメモリ上に保持し、クライアントの再接続または有効期限の経過まで維持します。クライアントが再接続しない場合、ノード上の切断セッション数はおおよそクライアント切断率に有効期限を掛けた値になります。

切断後のセッション保持は永続セッションの意図された動作です。ワークロードに応じて十分なメモリを割り当てるか、セッション状態をディスクに保存する durable sessions の利用を検討してください。

TIP

MQTT 設定をダッシュボード経由で行うには、ダッシュボード左ナビゲーションメニューの Management -> MQTT Settings をクリックしてください。ダッシュボードで設定した場合、その設定は設定ファイル内の同じ項目より優先されます。
設定ファイルから MQTT を設定する場合は、emqx.conf ではなく base.hocon の使用を推奨します。
emqx.conf に設定した場合、ダッシュボードでの変更は一時的なものとなり、EMQX の再起動時に失われるためです。

TIP

EMQX はカスタマイズニーズに対応するため、より多くの設定項目を提供しています。詳細は EMQX Enterprise Configuration Manual for Enterprise を参照してください。

ロードバランサーの設定

パフォーマンスとチューニング

ネットワークと TLS

認証

パスワード認証

認可

シングルサインオン (SSO)

監視

アクセスコントロール

統合

管理

Prometheus との統合

OpenTelemetry との統合

エンドツーエンドトレース

プラグインカタログ

MQTT 設定

基本的な MQTT 設定

厳格な MQTT パケット検証

サブスクリプション設定

遅延パブリッシュ設定

キープアライブ設定

動的キープアライブ調整

単一クライアント更新：`$SETOPTS/mqtt/keepalive`

一括更新：`$SETOPTS/mqtt/keepalive-bulk`

アクセス制御

セッション設定

パスワード認証

エンドツーエンドトレース

MQTT 設定 ​

基本的な MQTT 設定 ​

厳格な MQTT パケット検証 ​

サブスクリプション設定 ​

遅延パブリッシュ設定 ​

キープアライブ設定 ​

動的キープアライブ調整 ​

単一クライアント更新：$SETOPTS/mqtt/keepalive ​

一括更新：$SETOPTS/mqtt/keepalive-bulk ​

アクセス制御 ​

セッション設定 ​

MQTT 設定

基本的な MQTT 設定

厳格な MQTT パケット検証

サブスクリプション設定

遅延パブリッシュ設定

キープアライブ設定

動的キープアライブ調整

単一クライアント更新：`$SETOPTS/mqtt/keepalive`

一括更新：`$SETOPTS/mqtt/keepalive-bulk`

アクセス制御

セッション設定