MQTTとは｜仕組み・QoS・Retain・WillをPythonとパケットで解説

MQTT では、データを送る側（パブリッシャー）は「トピック」という宛名を付けてブローカーに送るだけです。ブローカーが、そのトピックを購読している側（サブスクライバー）全員に配ります。

flowchart TD
    P[パブリッシャー<br>温度センサー] -->|PUBLISH<br>home/living/temp| B[ブローカー<br>Mosquitto など]
    B -->|購読している人へ配る| S1[サブスクライバー<br>スマホのアプリ]
    B --> S2[サブスクライバー<br>記録するサーバー]
    S1 -.->|SUBSCRIBE<br>home/+/temp| B

急いでいる人向けに、よく調べられることの答えを先にまとめます。

知りたいこと	答え	詳しい節
MQTT とは	Pub/Sub 型の軽いメッセージのプロトコル	MQTTとは
読み方	エムキューティーティー	MQTTとは
ブローカー	メッセージを受けて購読者に配る中継サーバー	仕組み
QoS 0・1・2	最大1回・最低1回・ちょうど1回	QoS
Retain	最後の1件をブローカーが保存し、後から来た人に渡す	Retain
Will	異常切断のときにブローカーが代わりに送る	Will
ポート番号	1883（平文）・8883（TLS）	ポートとセキュリティ
HTTP との違い	つなぎっぱなしで双方向。ヘッダーが2バイトから	HTTPとの違い

確認環境：Ubuntu 24.04.4 LTS。ブローカーは apt で入れた Eclipse Mosquitto 2.0.18（mosquitto_pub・mosquitto_sub も同じ版）を、ネットワーク名前空間（ip netns）の中の 127.0.0.1 で動かしました。クライアントは Python 3.11.15 と paho-mqtt 2.1.0、パケットは tcpdump 4.99.4 で取り、MQTT の部分は標準ライブラリで書いた解析スクリプトで読みました。TLS は OpenSSL 3.0.13 で作った試験用の証明書で、比べるための HTTP は curl 8.5.0 で送りました。仕様は OASIS の MQTT Version 5.0・3.1.1、mqtt.org、IANA のポート番号の登録、Mosquitto・paho-mqtt・AWS IoT Core・EMQX・HiveMQ の公式ドキュメントで確かめました（確認日 2026年10月23日）。

MQTTとは（読み方・意味・使われる場面）

この節では、MQTT が何のためのプロトコルで、どんな場面で使われているのかが分かります。

MQTT（エムキューティーティー）は、IoT（モノのインターネット）のための、OASIS が標準にしたメッセージングのプロトコルです。 mqtt.org は、小さなプログラムと少ない通信量で遠くの機器をつなぐ、軽いパブリッシュ／サブスクライブ型の通信方式と説明しています。TCP の上で動き、HTTP と同じ「アプリケーション層」のプロトコルです。

AWS の解説によると、MQTT は1999年に石油・ガスのパイプラインを人工衛星の回線で監視するために生まれました。もとは Message Queuing Telemetry Transport の略でしたが、今は略語ではなく正式な名前として扱われています。

年	できごと
1999年	石油・ガス業界向けに発明される
2010年	IBM が MQTT 3.1 を誰でも実装できる形で公開
2013年	標準化のため OASIS に提出
2016年	MQTT 3.1.1 が ISO/IEC 20922:2016 になる
2019年	OASIS が MQTT 5.0 を公開

使われる場面の例です。

• センサーの値を集める（温度・湿度・電力。M5Stack・ESP32・ラズパイから送る）
• 工場の設備の状態や警報を集める
• スマートホームの機器を動かす（照明のオン・オフの命令を送る）
• スマホのアプリへの通知、チャット
• クラウドの IoT サービス（AWS IoT Core など）と機器をつなぐ

mqtt.org は、自動車・製造・通信・石油とガスなどの業界で使われている、と紹介しています。

MQTTの仕組み（ブローカー・パブリッシュ・サブスクライブ・トピック）

この節では、MQTT に出てくる4つの言葉と、メッセージがどう流れるのかが分かります。

用語	意味	例
クライアント	MQTT でつなぐプログラムや機器。送る側にも受ける側にもなる	センサー、スマホのアプリ
ブローカー	メッセージを受け取り、購読者に配る中継サーバー	Mosquitto、EMQX、AWS IoT Core
パブリッシュ（発行）	トピックを付けてメッセージをブローカーに送る	home/living/temperature に 23.5
サブスクライブ（購読）	受け取りたいトピックをブローカーに登録する	home/+/temperature を購読
トピック	/ で区切った階層の宛名。事前の登録は要らない	factory/line1/alarm

送る側と受ける側は、直接はつながりません。 どちらもブローカーにだけ TCP でつなぎ、つないだままにします。AWS の解説は、この「互いを知らない（空間）」「同時に動いていなくてよい（時間）」「待たずに処理を続けられる（同期）」の3つの切り離しを、MQTT の原則として挙げています。

メッセージが流れる順番です。

MQTT 5.0 の仕様は、制御パケットを15種類決めています（1バイト目の上位4ビットが種類）。

値	パケット	役割
1・2	CONNECT・CONNACK	接続の要求と返事
3	PUBLISH	メッセージを運ぶ
4〜7	PUBACK・PUBREC・PUBREL・PUBCOMP	QoS 1・2 の確認
8〜11	SUBSCRIBE・SUBACK・UNSUBSCRIBE・UNSUBACK	購読の登録と解除
12・13	PINGREQ・PINGRESP	生きていることの確認
14	DISCONNECT	切断
15	AUTH	認証のやり取り（MQTT 5 で追加）

MQTTを動かしてみる（Mosquittoのインストールとpub・sub）

この節では、ブローカーの Mosquitto を入れて、コマンドだけでメッセージを送って受け取るまでができます。

Eclipse Mosquitto は、オープンソースの MQTT ブローカーです。 公式サイトでは、EPL/EDL のライセンスで、MQTT 5.0・3.1.1・3.1 に対応し、小さなシングルボードコンピューターからサーバーまで動く軽さとされています。2026年10月23日時点のダウンロードページの最新版は 2.1.2 です。

OS	入れ方（公式のダウンロードページより）
Windows	mosquitto-2.1.2-install-windows-x64.exe を実行
Mac	Homebrew で brew install mosquitto
Ubuntu・Debian	sudo apt install mosquitto mosquitto-clients（新しい版は公式の PPA）

Ubuntu 24.04 の apt で入る版は 2.0.18 でした。ブローカー（mosquitto）とコマンドのクライアント（mosquitto_pub・mosquitto_sub）は、別のパッケージです。

sudo apt install mosquitto mosquitto-clients
mosquitto -h | head -3

mosquitto version 2.0.18

mosquitto is an MQTT v5.0/v3.1.1/v3.1 broker.

設定ファイルなしで起動すると「ローカルだけ」になる

設定ファイルを渡さずに mosquitto -v で起動したときのログです。

mosquitto version 2.0.18 starting
Using default config.
Starting in local only mode. Connections will only be possible from clients running on this machine.
Create a configuration file which defines a listener to allow remote access.
For more details see https://mosquitto.org/documentation/authentication-methods/
Opening ipv4 listen socket on port 1883.
Opening ipv6 listen socket on port 1883.
mosquitto version 2.0.18 running

Mosquitto 2.0.18 は、設定ファイルにリスナー（待ち受けるポート）を書かないと、同じマシンからしかつなげませんでした。 公式の mosquitto.conf の説明では、allow_anonymous（ユーザー名なしの接続を許すか）の既定は false で、リスナーが1つも無いときだけ true になり、ローカルからの接続だけを許します。ほかの PC やマイコンからつなぐときは、次のように書きます。

# /etc/mosquitto/conf.d/local.conf
listener 1883
allow_anonymous false
password_file /etc/mosquitto/passwd

この記事の試験では、ブローカーに次の設定を渡しました（ポート番号は記事用に変えています）。

per_listener_settings true
persistence false
log_type all
connection_messages true

# 1) 認証なし（試験用）
listener 36500 127.0.0.1
allow_anonymous true

# 2) WebSocket
listener 36501 127.0.0.1
protocol websockets
allow_anonymous true

# 3) ユーザー名とパスワードが必要
listener 36503 127.0.0.1
password_file /etc/mosquitto/passwd

# 4) allow_anonymous を書かない（既定 false）
listener 36504 127.0.0.1

mosquitto_sub で購読し、mosquitto_pub で発行する

ターミナルを2つ開き、1つ目で購読して待ち、2つ目から発行します。-v はトピック名も表示、-W 3 は3秒で終わるオプションです。

# ターミナル1：購読（2つ）
mosquitto_sub -h 127.0.0.1 -p 36500 -t 'home/+/temperature' -v -W 3
mosquitto_sub -h 127.0.0.1 -p 36500 -t 'home/#' -v -W 3

# ターミナル2：発行
mosquitto_pub -h 127.0.0.1 -p 36500 -t home/living/temperature -m 23.5
mosquitto_pub -h 127.0.0.1 -p 36500 -t home/kitchen/temperature -m 25.1
mosquitto_pub -h 127.0.0.1 -p 36500 -t home/living/humidity -m 48
mosquitto_pub -h 127.0.0.1 -p 36500 -t office/living/temperature -m 20.0

--- sub home/+/temperature
home/living/temperature 23.5
home/kitchen/temperature 25.1
Timed out
--- sub home/#
home/living/temperature 23.5
home/kitchen/temperature 25.1
home/living/humidity 48
Timed out

同じ4件の発行でも、購読しているトピックによって届くものが変わりました。 home/+/temperature には温度の2件だけ、home/# には home の下の3件が届き、office/... はどちらにも届いていません。ブローカーのログ（log_type all）には、受け取りと配った先が1行ずつ出ます。

New client connected from 127.0.0.1:45992 as auto-AE05894D-... (p2, c1, k60).
Sending CONNACK to auto-AE05894D-... (0, 0)
Received PUBLISH from auto-AE05894D-... (d0, q0, r0, m0, 'home/kitchen/temperature', ... (4 bytes))
Sending PUBLISH to auto-B6E926B3-... (d0, q0, r0, m0, 'home/kitchen/temperature', ... (4 bytes))
Sending PUBLISH to auto-4FCD12D8-... (d0, q0, r0, m0, 'home/kitchen/temperature', ... (4 bytes))

ログの p2 はプロトコルの版（MQTT 3.1.1）、c1 はクリーンセッション、k60 は Keep Alive 60秒、q0 は QoS、r0 は Retain です。クライアント ID を指定しないと、mosquitto_pub は auto-… という ID を自分で作ります。

mosquitto_pub・mosquitto_sub のよく使うオプション

オプション	意味
-h -p	ブローカーのホストとポート（既定 localhost・1883）
-t	トピック（sub はワイルドカード可・複数指定可）
-m	送るメッセージ。-f ファイル・-n（空）・-l（標準入力の行ごと）もある
-q	QoS（0・1・2）
-r	Retain 付きで発行
-i -c	クライアント ID の指定、クリーンセッションを使わない（sub）
-u -P	ユーザー名とパスワード
-V mqttv5	MQTT 5 で接続（既定は 3.1.1）
--cafile	TLS で使う CA 証明書
-v -F	トピック名も表示、表示の形を指定（sub）
-W -C	指定秒で終了、指定件数を受けたら終了（sub）
-d	送受信したパケットを表示（デバッグ）

PythonでMQTTを使う（paho-mqtt 2.x）

この節では、Python から購読と発行をするコピペで動くコードと、paho-mqtt 2.x の書き方の注意が分かります。

Python では Eclipse Paho の paho-mqtt を使います。 PyPI では 2026年10月23日時点の最新版が 2.1.0 で、MQTT 5.0・3.1.1・3.1 に対応しています。

pip install paho-mqtt

購読して受け取ったメッセージを表示し、QoS 0・1・2 で1回ずつ発行するコードです。

"""QoS 0・1・2 で1回ずつ発行し、QoS 2 で購読しているクライアントに届ける"""
import time
import paho.mqtt.client as mqtt

HOST, PORT = "127.0.0.1", 36500
sub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sub-1")
sub.on_message = lambda c, u, m: print(f"受信: {m.topic} {m.payload.decode()} qos={m.qos}")
sub.connect(HOST, PORT, keepalive=60)
sub.subscribe("s365/q", qos=2)
sub.loop_start(); time.sleep(0.5)            # 受信は別スレッドで回す

pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="pub-1")
pub.connect(HOST, PORT, keepalive=60); pub.loop_start(); time.sleep(0.3)
for q in (0, 1, 2):
    info = pub.publish("s365/q", f"q{q}", qos=q)
    info.wait_for_publish()                   # QoS 1・2 は確認の返事まで待つ
    print(f"発行: qos={q} mid={info.mid} rc={info.rc}")
    time.sleep(0.5)
pub.loop_stop(); pub.disconnect()
time.sleep(0.3)
sub.loop_stop(); sub.disconnect()

発行: qos=0 mid=1 rc=0
受信: s365/q q0 qos=0
受信: s365/q q1 qos=1
発行: qos=1 mid=2 rc=0
発行: qos=2 mid=3 rc=0
受信: s365/q q2 qos=2

paho-mqtt 2.x では、mqtt.Client() の最初の引数に CallbackAPIVersion.VERSION2 を渡します。 公式の移行ガイドによると、2.0 でコールバックの引数が変わり、on_connect は (client, userdata, flags, reason_code, properties) の5つになりました。引数を省くと、2.1.0 では次の警告が出て、1.x と同じ形のコールバック（VERSION1）として動きました。

DeprecationWarning: Callback API version 1 is deprecated, update to latest version
  mqtt.Client()

コールバック	VERSION2 の引数	1.x からの変化
on_connect	client, userdata, flags, reason_code, properties	rc が ReasonCode に
on_message	client, userdata, message	変わらない
on_subscribe	client, userdata, mid, reason_codes, properties	granted_qos が reason_codes に
on_publish	client, userdata, mid, reason_codes, properties	2つ増えた
on_disconnect	client, userdata, flags, reason_code, properties	flags・properties が増えた

loop_start() は受信の処理を別のスレッドで回し、loop_forever() は今のスレッドで回し続けます。受けるだけのプログラムなら loop_forever()、ほかの処理をしながら送るなら loop_start() を使います。

トピックの設計とワイルドカード（+・#）

この節では、トピックの書き方と、+・# がどのトピックに一致するのかが、実際の購読の結果で分かります。

トピックは / で区切った階層の文字列で、ブローカーに事前の登録は要りません。 Mosquitto の説明では、+ は1つの階層だけに、# は残りすべての階層に一致し、# はフィルターの最後に置きます。

6つのフィルターでそれぞれ購読し、7つのトピックに1回ずつ発行して、どれが届いたかを調べました。

"""トピックフィルター × トピックの一致を、実際に購読・発行して調べる"""
import time
import paho.mqtt.client as mqtt

HOST, PORT = "127.0.0.1", 36500
filters = ["home/+/temperature", "home/#", "home/+", "+/+/temperature", "#", "$SYS/broker/version"]
topics = ["home/living/temperature", "home/kitchen/temperature", "home/living",
          "home", "home/living/temperature/max", "office/living/temperature", "/home/living"]
got = {f: [] for f in filters}
subs = []
for f in filters:
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    c.on_message = lambda cl, ud, m, f=f: got[f].append(m.topic)
    c.connect(HOST, PORT); c.subscribe(f, qos=0); c.loop_start()
    subs.append(c)
time.sleep(0.5)
pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
pub.connect(HOST, PORT); pub.loop_start()
for t in topics:
    pub.publish(t, "x", qos=1).wait_for_publish()
time.sleep(1)
for f in filters:
    print(f"{f:22} -> {sorted(set(got[f]))}")

home/+/temperature     -> ['home/kitchen/temperature', 'home/living/temperature']
home/#                 -> ['home', 'home/kitchen/temperature', 'home/living', 'home/living/temperature', 'home/living/temperature/max']
home/+                 -> ['home/living']
+/+/temperature        -> ['home/kitchen/temperature', 'home/living/temperature', 'office/living/temperature']
#                      -> ['/home/living', 'home', 'home/kitchen/temperature', 'home/living', 'home/living/temperature', 'home/living/temperature/max', 'office/living/temperature']
$SYS/broker/version    -> ['$SYS/broker/version']

フィルター	届いたもの	届かなかったもの
home/+/temperature	living・kitchen の temperature	home/living/temperature/max（階層が深い）
home/#	home そのものと、下のすべて	office/...・/home/living
home/+	home/living だけ	home（階層が足りない）
+/+/temperature	home・office の両方の temperature	3階層でないもの
#	自分で発行した7件すべて	$SYS/...（ブローカーの情報）
$SYS/broker/version	ブローカーの版（ブローカーが保存しているもの）	―

読み取れることです。

• home/# は home そのものにも一致しました。 # は「0個以上の階層」です。
• /home/living は先頭が空の階層になり、home/# には一致しません。 HiveMQ の解説も、先頭の / は空の階層を増やすだけなので付けないことをすすめています。
• # で全部を購読しても $SYS/ で始まるトピックは届きませんでした。 $SYS/ は Mosquitto が自分の状態を出す場所です。HiveMQ の解説も、$ で始まるトピックはブローカーの内部用で、# の購読には含まれず、クライアントからは発行しないものとしています。$SYS/broker/version のように直接指定すると届きます。Mosquitto は sys_interval（既定10秒）ごとに更新します。

ワイルドカードの書き間違いは、コマンドでも Python でもエラーになります。

$ mosquitto_sub -t 'home/#/x'
Error: Invalid subscription topic 'home/#/x', are all '+' and '#' wildcards correct?

$ mosquitto_sub -t 'home/te+'
Error: Invalid subscription topic 'home/te+', are all '+' and '#' wildcards correct?

$ mosquitto_pub -t 'home/+/temperature' -m 1
Error: Invalid publish topic 'home/+/temperature', does it contain '+' or '#'?

ValueError: Invalid subscription filter.            # client.subscribe("a/#/b")
ValueError: Publish topic cannot contain wildcards.  # client.publish("a/+/b", "x")

ワイルドカードは購読だけで使え、発行のトピックには使えません。 + と # は1つの階層をまるごと占める必要があり、te+ のように文字と混ぜられません。

トピックの決め方の目安

• 1大きい区分から小さい区分へ並べる場所/建物/設備/計測値 の順にすると、+ と # で範囲を選びやすい
• 2先頭に / を付けない先頭が空の階層になり、home/# に一致しなくなった
• 3空白・日本語・大文字の混在を避けるトピックは大文字と小文字を区別する（HiveMQ の解説）。Home と home は別のトピック
• 4機器の ID を階層に入れるsensors/m5stack-01/temperature のようにすると、1台だけ・全部を選び分けられる
• 5$ で始めない$SYS などブローカーが使う名前と重なる

QoS 0・1・2の違い（パケットをtcpdumpで見る）

この節では、QoS の3つのレベルで何が違うのかを、実際に流れたパケットの数と大きさ、速さで確かめられます。

QoS（Quality of Service）は、メッセージを届ける確かさのレベルです。 mqtt.org は、0 を「最大1回（at most once）」、1 を「最低1回（at least once）」、2 を「ちょうど1回（exactly once）」と説明しています。

QoS	届き方	確認のパケット	向いている用途
0	最大1回。届かないこともある	なし	毎秒送る温度など、次の値がすぐ来るもの
1	最低1回。重なって届くことがある	PUBACK	警報・ログ。受け側で重なりを扱えるもの
2	ちょうど1回	PUBREC・PUBREL・PUBCOMP	回数を数える・課金など、重なると困るもの

実際に流れたパケット

前の節の Python のコード（QoS 0・1・2 で1回ずつ発行）を動かしながら、名前空間の中で tcpdump -i lo -w q.pcap 'tcp port 36500' で取り、MQTT のパケットを1つずつ取り出しました。54007 番が購読側、34769 番が発行側の接続です。

"""pcap（tcpdump -w）から MQTT の制御パケットを1つずつ取り出して表示する（標準ライブラリだけ）"""
import struct, sys
NAMES = {1:"CONNECT",2:"CONNACK",3:"PUBLISH",4:"PUBACK",5:"PUBREC",6:"PUBREL",7:"PUBCOMP",
         8:"SUBSCRIBE",9:"SUBACK",10:"UNSUBSCRIBE",11:"UNSUBACK",12:"PINGREQ",13:"PINGRESP",14:"DISCONNECT",15:"AUTH"}
BROKER = int(sys.argv[2])
def remlen(b, i):                     # 残りの長さ（1〜4バイトの可変長の整数）
    mul, val = 1, 0
    while True:
        d = b[i]; i += 1; val += (d & 127) * mul; mul *= 128
        if d < 128: return val, i
data = open(sys.argv[1], "rb").read()
link = struct.unpack("<I", data[20:24])[0]; off = 24; t0 = None
while off + 16 <= len(data):
    ts, tus, incl, _ = struct.unpack("<IIII", data[off:off+16]); pkt = data[off+16:off+16+incl]; off += 16 + incl
    t = ts + tus / 1e6; t0 = t0 or t
    ip = pkt[14:] if link == 1 else pkt[16:]
    tcp = ip[(ip[0] & 15) * 4:]
    sp, dp = struct.unpack(">HH", tcp[:4]); pl = tcp[(tcp[12] >> 4) * 4:]
    if not pl: continue
    arrow = f"broker -> {dp}" if sp == BROKER else f"{sp} -> broker"
    i = 0
    while i < len(pl):
        h = pl[i]; typ = h >> 4; ln, j = remlen(pl, i + 1); body = pl[j:j+ln]; extra = ""
        if typ == 3:
            q = (h >> 1) & 3; tl = struct.unpack(">H", body[:2])[0]; topic = body[2:2+tl].decode()
            pid = struct.unpack(">H", body[2+tl:4+tl])[0] if q else None
            extra = f"qos={q} retain={h & 1} dup={h >> 3 & 1} topic={topic}" + (f" id={pid}" if pid else "") + f" payload={body[2+tl+(2 if q else 0):]!r}"
        elif typ in (4, 5, 6, 7): extra = f"id={struct.unpack('>H', body[:2])[0]}"
        print(f"{t - t0:7.4f}  {arrow:16} {NAMES[typ]:10} {1 + (j - i - 1) + ln:3}B  {extra}")
        i = j + ln

 0.0007  54007 -> broker  CONNECT     19B
 0.0007  broker -> 54007  CONNACK      4B
 0.0008  54007 -> broker  SUBSCRIBE   13B
 0.0008  broker -> 54007  SUBACK       5B
 0.5016  34769 -> broker  CONNECT     19B
 0.5018  broker -> 34769  CONNACK      4B
 0.8028  34769 -> broker  PUBLISH     12B  qos=0 retain=0 dup=0 topic=s365/q payload=b'q0'
 0.8029  broker -> 54007  PUBLISH     12B  qos=0 retain=0 dup=0 topic=s365/q payload=b'q0'
 1.3033  34769 -> broker  PUBLISH     14B  qos=1 retain=0 dup=0 topic=s365/q id=2 payload=b'q1'
 1.3035  broker -> 54007  PUBLISH     14B  qos=1 retain=0 dup=0 topic=s365/q id=1 payload=b'q1'
 1.3035  broker -> 34769  PUBACK       4B  id=2
 1.3037  54007 -> broker  PUBACK       4B  id=1
 1.8042  34769 -> broker  PUBLISH     14B  qos=2 retain=0 dup=0 topic=s365/q id=3 payload=b'q2'
 1.8043  broker -> 34769  PUBREC       4B  id=3
 1.8045  34769 -> broker  PUBREL       4B  id=3
 1.8045  broker -> 54007  PUBLISH     14B  qos=2 retain=0 dup=0 topic=s365/q id=2 payload=b'q2'
 1.8045  broker -> 34769  PUBCOMP      4B  id=3
 1.8047  54007 -> broker  PUBREC       4B  id=2
 1.8048  broker -> 54007  PUBREL       4B  id=2
 1.8050  54007 -> broker  PUBCOMP      4B  id=2
 2.8057  34769 -> broker  DISCONNECT   2B
 3.8194  54007 -> broker  DISCONNECT   2B

1つの区間（送る側とブローカーのあいだ）でのやり取りを図にすると、次のとおりです。ブローカーと購読側のあいだでも、同じやり取りがもう一度行われます。

sequenceDiagram
    participant C as 送る側
    participant B as 受ける側
    Note over C,B: QoS 0
    C->>B: PUBLISH
    Note over C,B: QoS 1
    C->>B: PUBLISH id=2
    B->>C: PUBACK id=2
    Note over C,B: QoS 2
    C->>B: PUBLISH id=3
    B->>C: PUBREC id=3
    C->>B: PUBREL id=3
    B->>C: PUBCOMP id=3

読み取れることです。

• QoS 1・2 の PUBLISH には2バイトのパケット ID が付き、QoS 0 より2バイト大きくなりました。 確認のパケット（PUBACK など）は、同じ ID を返して「どれを受け取ったか」を伝えます。
• パケット ID は区間ごとに別です。 発行側からは id=2 で届いた q1 を、ブローカーは購読側へ id=1 で送りました。ID は「つないでいる2者のあいだ」だけで使う番号です。
• QoS 2 では、ブローカーは PUBREL を受けてから PUBCOMP を返し、購読側への配送を始めています。 4回のやり取りで「同じメッセージを2回処理しない」ことを確かめます。

発行の QoS と購読の QoS

購読側が受け取る QoS は、発行の QoS と購読の QoS の低いほうになります。 Mosquitto の説明もこの決まりを書いています。上の例は購読を QoS 2 にしたので、q0・q1・q2 はそれぞれ QoS 0・1・2 のまま届きました。購読を QoS 0 にすると、QoS 2 で発行しても購読側には QoS 0 で届きます。

PUBLISH の12バイトを読む

QoS 0 の PUBLISH を tcpdump -X で見ると、TCP のヘッダーの後ろの12バイトが MQTT です。

IP 127.0.0.1.34769 > 127.0.0.1.36500: Flags [P.], seq 1982874163:1982874175, ack 2273476176, win 64, options [nop,nop,TS val 2867062457 ecr 650164975], length 12
    0x0000:  4500 0040 7957 4000 4006 c35e 7f00 0001  E..@yW@.@..^....
    0x0010:  7f00 0001 87d1 8e94 7630 4233 8782 7e50  ........v0B3..~P
    0x0020:  8018 0040 fe34 0000 0101 080a aae3 e6b9  ...@.4..........
    0x0030:  26c0 baef 300a 0006 7333 3635 2f71 7130  &...0...s365/qq0

バイト	値	意味
1	30	上位4ビット 3＝PUBLISH、下位4ビット 0＝QoS 0・Retain なし
2	0a	残りの長さ 10 バイト
3〜4	0006	トピック名の長さ 6
5〜10	7333 3635 2f71	トピック名 s365/q
11〜12	7130	ペイロード q0

MQTT のヘッダーは、種類とフラグの1バイトと、残りの長さの1〜4バイトだけです。 残りの長さは可変長の整数で、MQTT 5.0 の仕様では最大4バイト、268,435,455（約256MB）まで表せます。購読側の CONNECT（19バイト）も同じように読めます。

10 11 | 0004 4d515454 | 04 | 02 | 003c | 0005 7375622d31
CONNECT 残り17 | "MQTT" | 版4＝3.1.1 | クリーンセッション | Keep Alive 60秒 | ID "sub-1"

QoS ごとの速さ（2,000件を送る）

同じマシンの中（127.0.0.1）で、QoS ごとに16バイトのメッセージを2,000件発行し、購読側に全部届くまでの時間を5回測った中央値です。

QoS 1 は QoS 0 の約4分の1、QoS 2 は約5分の1の速さでした。 確認の往復が増えるほど遅くなります。ネットワークの遅れがある本物の回線では、往復の回数の差がさらに大きく効きます。

QoS 1 でも、ブローカーのキューがあふれると捨てられます。 既定の設定のブローカーで同じ測定をしたところ、QoS 1 の6回のうち4回で 1,629〜1,877 件しか届かず、ブローカーのログに次の行が出ました。

Outgoing messages are being dropped for client bs1.

Mosquitto の max_queued_messages（1クライアントあたり、送信中の分を除いてためておく QoS 1・2 のメッセージの数）の既定は1000です。受け取る側の処理が追いつかず、ためた数が上限を超えたためです。max_queued_messages 10000 にしたブローカーでは、5回とも全件が届き、このログは出ませんでした。QoS は「ブローカーと1区間のあいだの確認」であり、ブローカーの上限を超えた分まで守るものではありません。

QoS の選び方

flowchart TD
    A{1件欠けても困らない?<br>次の値がすぐ来る} -->|はい| Q0[QoS 0<br>毎秒の温度]
    A -->|いいえ| C{2回届いても<br>困らない?}
    C -->|はい| Q1[QoS 1<br>警報・ログ]
    C -->|いいえ| Q2[QoS 2<br>回数・課金]

QoS 1 で重なって届いても困らないように、受け取る側でメッセージに付けた番号（連番や ID）を見て、同じものを2回処理しない作りにしておく方法もよく使われます。AWS IoT Core のように QoS 2 に対応していないブローカーもあります（後の節の表）。

Retain・Will・Keep Aliveの使い方

この節では、MQTT のよく使う3つの機能が、それぞれ何をして、どう書けばよいのかが実際の動きで分かります。

Retain：後から購読した人に最後の値を渡す

mosquitto_pub -h 127.0.0.1 -p 36500 -t home/living/temperature -m 23.5 -r   # retain 付き
mosquitto_pub -h 127.0.0.1 -p 36500 -t home/living/humidity -m 48           # retain なし
# 発行した後で購読する（%r は retain フラグ）
mosquitto_sub -h 127.0.0.1 -p 36500 -t 'home/living/#' -F '%r %t %p' -W 1
# 空のメッセージを retain 付きで送ると、保存されたものが消える
mosquitto_pub -h 127.0.0.1 -p 36500 -t home/living/temperature -r -n
mosquitto_sub -h 127.0.0.1 -p 36500 -t 'home/living/#' -F '%r %t %p' -W 1

1 home/living/temperature 23.5
Timed out
Timed out

発行した後に購読しても、retain 付きの温度だけが retain フラグ 1 で届き、retain なしの湿度は届きませんでした。 MQTT 3.1.1 の仕様では、retain 付きで本文0バイトのメッセージを送ると、そのトピックに保存されているメッセージは消えます。実際、消した後の購読では何も届きませんでした。

使いどころ	例
状態を表すトピック	機器の online・offline、スイッチのオン・オフ
たまにしか変わらない設定値	しきい値、ファームウェアの版
使わないほうがよいもの	毎秒の計測値の履歴（最後の1件しか残らない）

Will：異常に切れたら代わりに知らせる

Will は、接続のときに「トピック・メッセージ・QoS・Retain」をブローカーに預けておく機能です。Mosquitto の説明では、クライアントが予期せず切断したときに送られます。

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sensor-B")
c.will_set("status/sensor-B", "offline", qos=1, retain=True)   # connect の前に預ける
c.connect("127.0.0.1", 36500, keepalive=60)
c.loop_start()
c.publish("status/sensor-B", "online", qos=1, retain=True)       # つないだら online を保存

2つの切れ方を試しました。

• sensor-A：CONNECT（Keep Alive 2秒・Will は offline）を送った後、何も送らない（電源が落ちた・電波が切れたときの再現）
• sensor-B：上のコードでつなぎ、disconnect() で正しく切断する

購読していた監視用のクライアントの表示です。

 0.30s sensor-A 接続（keepalive=2秒）。CONNACK = 20020000
 4.82s 受信: status/sensor-A = offline (retain=False)
 4.82s ブローカーが接続を閉じた（recv=b''）
 5.62s 受信: status/sensor-B = online (retain=False)
 5.62s sensor-B は DISCONNECT で正常に切断

ブローカーのログです。

New client connected from 127.0.0.1:53922 as sensor-A (p2, c1, k2).
Will message specified (7 bytes) (r1, q1).
    status/sensor-A
Client sensor-A has exceeded timeout, disconnecting.
...
Received DISCONNECT from sensor-B
Client sensor-B disconnected.

応答の止まった sensor-A は、ブローカーが切断して Will の offline を発行しました。DISCONNECT で切った sensor-B の Will は発行されませんでした。 online を retain 付きで発行し、Will も retain 付きの offline にしておくと、後から購読した人にも「今つながっているか」が伝わります。実際、次に監視のクライアントをつなぐと、offline (retain=True) と online (retain=True) がすぐ届きました。

Keep Alive：何秒で「切れた」と判断されるか

クライアントは、Keep Alive の秒数のあいだに何も送るものが無いと PINGREQ を送ります。 Keep Alive を2秒にして何も発行せずに待ったときのパケットです。

 0.0001  49749 -> broker  CONNECT     23B
 0.0003  broker -> 49749  CONNACK      4B
 2.0022  49749 -> broker  PINGREQ      2B
 2.0023  broker -> 49749  PINGRESP     2B
 4.0048  49749 -> broker  PINGREQ      2B
 4.0048  broker -> 49749  PINGRESP     2B
 5.5008  49749 -> broker  DISCONNECT   2B

MQTT 5.0 の仕様では、Keep Alive が0でなく、その1.5倍の時間クライアントから何も届かないと、サーバー（ブローカー）は接続を閉じます。

Mosquitto 2.0.18 では、1.5倍の時間を過ぎてから少し後（2秒なら約4.5秒、10秒なら約17.1秒）に切断されました。 Keep Alive を短くすると切断に早く気づけますが、PINGREQ の通信と電池の消費が増えます。paho-mqtt 2.1.0 の connect() の既定は60秒、mosquitto_pub・mosquitto_sub も60秒（ログの k60）でした。

クリーンセッションと永続セッション

この節では、切断しているあいだのメッセージを、再接続したときに受け取る方法が分かります。

QoS 1 で購読してから切断し、そのあいだに3件発行して、同じ ID でつなぎ直しました（購読はし直しません）。

"""クリーンセッションと永続セッション：切断中の QoS 1 のメッセージは届くか"""
import time
import paho.mqtt.client as mqtt
HOST, PORT = "127.0.0.1", 36505

def run(clean):
    got = []
    cid = f"logger-clean{int(clean)}"
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=cid, clean_session=clean)
    c.on_message = lambda cl, u, m: got.append(m.payload.decode())
    c.connect(HOST, PORT); c.subscribe("factory/line1/alarm", qos=1); c.loop_start(); time.sleep(0.3)
    c.disconnect(); c.loop_stop()                        # 一度切断する

    p = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="plc")
    p.connect(HOST, PORT); p.loop_start()
    for i in range(1, 4):                                # 切断中に3件発行
        p.publish("factory/line1/alarm", f"alarm-{i}", qos=1).wait_for_publish()
    p.disconnect(); p.loop_stop()

    flags = {}
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=cid, clean_session=clean)
    c.on_message = lambda cl, u, m: got.append(m.payload.decode())
    c.on_connect = lambda cl, u, f, rc, pr: flags.update(sp=f.session_present)
    c.connect(HOST, PORT); c.loop_start(); time.sleep(1)  # 購読し直さずに待つ
    c.disconnect(); c.loop_stop()
    print(f"clean_session={clean}: session_present={flags['sp']} 受け取った={got}")

run(True)
run(False)

clean_session=True: session_present=False 受け取った=[]
clean_session=False: session_present=True 受け取った=['alarm-1', 'alarm-2', 'alarm-3']

永続セッションでは、CONNACK の session_present が True になり、切断中の3件がつなぎ直した直後に届きました。 クリーンセッションでは何も届きません。永続セッションには、同じクライアント ID・QoS 1 以上での購読と発行の両方が要ります。

Mosquitto 2.0.18 で per_listener_settings true にした設定のブローカーでは、同じ手順で session_present は True なのに、3件は届きませんでした。 per_listener_settings を書かない設定のブローカーに変えると、上のとおり3件届きました。リスナーごとに認証を分ける設定と永続セッションを組み合わせるときは、切断中のメッセージが届くかを先に試しておくと安全です。

MQTT 5 では、クリーンセッションは「Clean Start」と「Session Expiry Interval（切断後にセッションを残す秒数）」の2つに分かれました。HiveMQ の解説では、Session Expiry Interval を0にすると、接続が閉じたときにセッションが消えます。

MQTTとHTTP・WebSocketの違い

この節では、MQTT と HTTP の違いと、どちらを選べばよいのかが、同じデータを送ったバイト数で分かります。

HTTP は「頼まれたら返す」要求と応答の形、MQTT は「つないだままで、どちらからでも送れる」発行と購読の形です。 AWS の解説も、MQTT は RESTful ではなく、つないだままの TCP の上でブローカーが押し出す（push）形だと説明しています。

同じ q0（2バイト）を、HTTP の POST と MQTT の PUBLISH で送りました。

POST /s365/q HTTP/1.1
Host: 127.0.0.1:36502
User-Agent: curl/8.5.0
Accept: */*
Content-Type: text/plain
Content-Length: 2

q0
request bytes: 132 | header: 130 | body: 2 | response bytes: 41

比べる点	MQTT	HTTP
通信の形	発行と購読（1対多）	要求と応答（1対1）
接続	つないだまま	要求ごと（keep-alive で使い回しもできる）
サーバーから送る	できる（購読者へ push）	基本はできない（ポーリング・SSE など）
ヘッダーの大きさ	2バイトから	数十〜数百バイト
届け方の保証	QoS 0・1・2	TCP まかせ。再送はアプリで
切断の通知	Will・Keep Alive	仕組みは無い
向いているもの	センサー・通知・機器の制御	Web ページ・REST API
中継	ブローカーが要る	直接つなげる

ブラウザから MQTT を使うときは、MQTT over WebSocket を使います。 ブラウザの JavaScript は TCP のソケットを直接開けないためです。Mosquitto はリスナーに protocol websockets を書くと WebSocket で待ち受けます。paho-mqtt で transport="websockets" を指定してつないだときの、最初の HTTP の要求です。

GET /mqtt HTTP/1.1
Host: 127.0.0.1:36501
Upgrade: websocket
Connection: Upgrade
Origin: http://127.0.0.1:36501
Sec-WebSocket-Key: BeSbkeFYRAqnxJLPCwZIcw==
Sec-Websocket-Version: 13
Sec-Websocket-Protocol: mqtt

WebSocket 経由で受信: [('ws/test', 'hello over websocket')]

Sec-Websocket-Protocol: mqtt で「この WebSocket の上で MQTT を話す」と伝え、その後は WebSocket のフレームの中に、同じ MQTT のパケットを入れて送ります。 WebSocket だけで作るチャットとの違いは、トピック・QoS・Retain をブローカーが受け持つ点です。

選ぶもの	向いている場面
MQTT（TCP）	マイコン・センサー・サーバーどうし。回線が細い・切れやすい
MQTT over WebSocket	ブラウザのダッシュボードで、機器と同じトピックを見る
HTTP（REST）	1回だけの要求と応答、Web API、ファイルの取得
WebSocket だけ	1つのサーバーと双方向にやり取りするアプリ

MQTTのポート番号とセキュリティ（1883・8883・TLS・認証）

この節では、MQTT のポート番号と、インターネット越しに使うときに最低限そろえる設定が分かります。

IANA には、MQTT が 1883 番、TLS を使う MQTT（secure-mqtt）が 8883 番で登録されています。 WebSocket の番号は決まっておらず、ブローカーやサービスごとに違います。

flowchart TD
    A[MQTT クライアント] -->|1883 平文| B[ブローカー]
    A -->|8883 TLS| B
    W[ブラウザ] -->|WebSocket<br>ws:// wss://| B

ユーザー名とパスワード

sudo mosquitto_passwd -c /etc/mosquitto/passwd sensor1     # ファイルを作り、パスワードを聞かれる

認証が要るリスナーに、いろいろな形でつないだ結果です。

== 認証が必要なリスナーに、パスワードなしで
Connection error: Connection Refused: not authorised.
exit=5
== 間違ったパスワード（MQTT 5）
Connection error: Not authorized
exit=135
== 正しいパスワード
Timed out
== allow_anonymous を書かないリスナー
Connection error: Connection Refused: not authorised.
exit=5

MQTT 3.1.1 の接続拒否は CONNACK の戻り値5（not authorised）、MQTT 5 は理由コード 135（0x87 Not authorized）として返りました。 MQTT 5.0 の仕様では、0x86 は Bad User Name or Password、0x87 は Not authorized です。Mosquitto 2.0.18 は、パスワード違いにも 0x87 を返しました。正しいパスワードの行の Timed out は、つながって -W 1 の1秒が過ぎたという意味です。

TLS（8883）で暗号化する

試験用の CA と、それで署名したサーバー証明書を作り、Mosquitto に設定しました。

listener 8883
cafile /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key

mosquitto_pub -h localhost -p 8883 --cafile ca.crt -t tls/test -m hello -d

Client null sending CONNECT
Client null received CONNACK (0)
Client null sending PUBLISH (d0, q0, r0, m1, 'tls/test', ... (5 bytes))
Client null sending DISCONNECT

Python では tls_set() を connect の前に呼びます。

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
c.tls_set(ca_certs="ca.crt")      # 省くと OS の信頼済みの CA で確かめる
c.connect("localhost", 8883)

試したこと	結果
CA 証明書を指定（コマンド）	接続できた（CONNACK 0）
OS の CA だけ（コマンド）	Error: Protocol error。ブローカーのログは tlsv1 alert unknown ca
OS の CA だけ（Python）	SSLCertVerificationError: … self-signed certificate in certificate chain
TLS のポートに平文でつなぐ	Error: The connection was lost.。ログは wrong version number

• 1インターネット越しは 8883（TLS）を使う1883 の平文では、パスワードもメッセージもそのまま流れる
• 2allow_anonymous false とパスワードかクライアント証明書allow_anonymous の既定は false。リスナーを書くと匿名は拒否される
• 3ACL で読み書きできるトピックを分けるセンサーは自分のトピックに発行だけ、など。Mosquitto は acl_file で書く
• 4クライアント ID を機器ごとに一意にする同じ ID が後からつなぐと、先の接続は切られる（エラーの節）
• 5パスワードファイルの持ち主をブローカーの利用者にするグループが違うと、2.0.18 は将来は読まないという警告を出した

MQTTブローカーの種類と選び方（Mosquitto・EMQX・HiveMQ・AWS IoT Core）

この節では、主なブローカーの特徴と、クラウドのサービスで仕様と違う点が分かります。

ブローカー	種類・ライセンス	対応する版	特徴（公式より）
Eclipse Mosquitto	オープンソース・EPL/EDL	5.0・3.1.1・3.1	軽く、小さなボードからサーバーまで動く
EMQX	5.9.0 から Business Source License 1.1（以前は Apache 2.0）	5.0・3.1.1・3.1	大規模向け。クラウド版もある
HiveMQ Community Edition	オープンソース・Apache 2.0	5.0・3.1.1・3.1	Java で動く（Java 11 以上）
AWS IoT Core	AWS のマネージドサービス	5.0・3.1.1	QoS 0・1 のみ。AWS の各サービスとつながる

2026年10月23日時点の各公式ページ（GitHub の README を含む）の内容です。

AWS IoT Core は、MQTT の仕様と違う点を公式ドキュメントにまとめています。 主なものは次のとおりです。

項目	AWS IoT Core のふるまい
QoS 2	対応していない。要求すると PUBACK・SUBACK を返さない
ワイルドカードの購読と Retain	購読した時点では、保存された Retain は届かない（完全一致なら届く）
同じクライアント ID	2つ同時にはつなげない。後の接続で前が切られる
セッションの有効期限	MQTT 3 は既定1時間。MQTT 5 は最大7日まで
メッセージと確認の順番	保証されない

試しに使うなら Mosquitto を手元で動かすのが手軽で、機器が多くなったら EMQX・HiveMQ などの大規模向け、AWS の各サービスにつなぐなら AWS IoT Core、という選び方がよく見られます。クライアントでは、画面でつないでトピックを見られる MQTTX・MQTT Explorer なども使われています。

MQTT 5の新機能

この節では、MQTT 5 で何が増えたのかと、いくつかの機能の実際の動きが分かります。

MQTT 5.0 は2019年に OASIS が公開した版で、エラーの理由を返す・大規模で使う・機能を広げる、ための仕組みが足されました。 仕様は、目的として「大規模なシステムに向けた改良」「エラーの報告の改善」「能力の確認や要求と応答の型の定式化」「ユーザープロパティなどの拡張の仕組み」「性能と小さなクライアントへの対応」を挙げています。

機能	何ができるか
Reason Code（理由コード）	CONNACK・PUBACK などで失敗の理由を返す
Clean Start・Session Expiry Interval	クリーンセッションを分け、切断後にセッションを残す秒数を決める
Message Expiry Interval	メッセージの有効期限。過ぎたら配らない
Topic Alias	長いトピック名の代わりに番号を送る
User Properties	パケットに任意のキーと値を付ける
Shared Subscriptions	$share/グループ/トピック で複数の購読者に振り分ける
Request/Response	Response Topic・Correlation Data で返事の宛先を伝える
Will Delay Interval	Will を発行するまで待つ秒数
AUTH パケット	チャレンジとレスポンスの認証、つないだままの再認証
サーバーからの DISCONNECT	ブローカーが理由を付けて切断を伝える

ユーザープロパティ・メッセージの有効期限・共有サブスクリプションを、paho-mqtt で protocol=mqtt.MQTTv5 にして試しました。

"""MQTT 5 の新機能を試す：ユーザープロパティ・メッセージの有効期限・共有サブスクリプション"""
import time
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes
HOST, PORT = "127.0.0.1", 36500

def client(cid, on_msg=None):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=cid, protocol=mqtt.MQTTv5)
    if on_msg: c.on_message = on_msg
    c.connect(HOST, PORT); c.loop_start(); return c

# 1) ユーザープロパティ・Content Type・応答先トピック
def show(c, u, m):
    p = m.properties
    print("受信:", m.topic, m.payload.decode(), "| ContentType:", getattr(p, "ContentType", None),
          "| UserProperty:", getattr(p, "UserProperty", None), "| ResponseTopic:", getattr(p, "ResponseTopic", None))
s = client("v5-sub", show); s.subscribe("v5/demo", qos=1); time.sleep(0.3)
pub = client("v5-pub")
pr = Properties(PacketTypes.PUBLISH)
pr.ContentType = "application/json"
pr.UserProperty = [("unit", "celsius"), ("device", "m5stack-01")]
pr.ResponseTopic = "v5/reply/m5stack-01"; pr.CorrelationData = b"req-42"
pub.publish("v5/demo", '{"t":23.5}', qos=1, properties=pr).wait_for_publish(); time.sleep(0.5)

# 2) メッセージの有効期限（Message Expiry Interval）
pe = Properties(PacketTypes.PUBLISH); pe.MessageExpiryInterval = 2
pub.publish("v5/expire/a", "期限2秒", qos=1, retain=True, properties=pe).wait_for_publish()
pub.publish("v5/expire/b", "期限なし", qos=1, retain=True).wait_for_publish()
time.sleep(3)
got = []
late = client("v5-late", lambda c, u, m: got.append((m.topic, m.payload.decode())))
late.subscribe("v5/expire/#", qos=1); time.sleep(0.5)
print("3秒後に購読して届いた retain:", got)

# 3) 共有サブスクリプション（$share/グループ名/トピック）
count = {"w1": [], "w2": []}
w1 = client("w1", lambda c, u, m: count["w1"].append(m.payload.decode())); w1.subscribe("$share/workers/jobs", qos=1)
w2 = client("w2", lambda c, u, m: count["w2"].append(m.payload.decode())); w2.subscribe("$share/workers/jobs", qos=1)
time.sleep(0.5)
for i in range(1, 7):
    pub.publish("jobs", f"job-{i}", qos=1).wait_for_publish()
time.sleep(0.5)
print("共有サブスクリプションの振り分け:", count)

受信: v5/demo {"t":23.5} | ContentType: application/json | UserProperty: [('unit', 'celsius'), ('device', 'm5stack-01')] | ResponseTopic: v5/reply/m5stack-01
3秒後に購読して届いた retain: [('v5/expire/b', '期限なし')]
共有サブスクリプションの振り分け: {'w1': ['job-1', 'job-3', 'job-5'], 'w2': ['job-2', 'job-4', 'job-6']}

• ユーザープロパティと Content Type は、本文とは別に受け取る側へ届きました。 単位や機器の ID を本文の JSON に入れずに渡せます。
• 有効期限2秒の Retain は、3秒後の購読では届かず、期限なしのものだけが届きました。 古くなった値を配らないために使います。
• 共有サブスクリプションでは、6件が2つの購読者に3件ずつ交互に配られました。 同じ処理をする複数のワーカーに仕事を分けるときに使います。普通の購読なら、2つとも6件を受け取ります。

Topic Alias で何バイト減るか

52文字のトピック factory/tokyo/building-a/line-3/robot-07/temperature に 23.5 を5回送り、PUBLISH の大きさを比べました。

送り方	PUBLISH の大きさ
エイリアスなし（2回）	61バイト
1回目：トピック名と番号1を対応づける	64バイト
2回目から：トピック名を空にして番号1だけ	12バイト

2回目からは61バイトが12バイトになりました。 ブローカーのログでは、5回とも元の長いトピック名で受け取ったと記録され、購読側にも元の名前で配られます。番号は1つの接続の中だけで有効です。

MQTTのよくあるエラーと直し方

この節では、MQTT で実際に出たエラー文から、原因と直し方を引けます。

エラー文・症状	原因	直し方
Error: Connection refused	ブローカーが動いていない・ポート違い	起動とポート番号（1883 など）を確かめる
ConnectionRefusedError: [Errno 111] Connection refused（Python）	同上	同上。ss -ltnp で待ち受けを見る
ほかの PC からつながらない／ログに Starting in local only mode.	設定ファイルにリスナーが無い	listener 1883 と認証の設定を書く
Connection error: Connection Refused: not authorised.（終了コード5）	匿名を許していないリスナー	-u・-P を付ける。試験だけなら allow_anonymous true
Connection error: Not authorized（MQTT 5、135）	ユーザー名かパスワードが違う	mosquitto_passwd で作り直す
Error: Invalid subscription topic '…', are all '+' and '#' wildcards correct?	# が最後でない・te+ のように混ぜた	+・# は1つの階層をまるごと使う
Error: Invalid publish topic '…', does it contain '+' or '#'?	発行にワイルドカードを使った	発行は具体的なトピック名にする
ValueError: Invalid subscription filter.・Publish topic cannot contain wildcards.	Python で同じ書き間違い	同上
rc = 4 The client is not currently connected.	connect の前・切断後に publish した	connect と loop_start の後に送る
ログに Client … already connected, closing old connection.	同じクライアント ID が2つ	機器ごとに一意の ID にする
ログに Client … has exceeded timeout, disconnecting.	Keep Alive の1.5倍、何も届かなかった	loop が止まっていないか、回線を確かめる
ログに Outgoing messages are being dropped for client …	受け取る側が遅く、max_queued_messages（既定1000）を超えた	受け側を速くする・上限を上げる
切断中のメッセージが届かない	クリーンセッション・QoS 0・ID が変わった	clean_session=False・QoS 1 以上・同じ ID
Error: Protocol error／ログに tlsv1 alert unknown ca	サーバー証明書の CA を知らない	--cafile・tls_set(ca_certs=…) を指定
SSLCertVerificationError: … self-signed certificate in certificate chain	Python で CA を指定していない	同上
Error: The connection was lost.／ログに wrong version number	TLS のポートに平文でつないだ	TLS の設定を付けるか 1883 につなぐ
DeprecationWarning: Callback API version 1 is deprecated	paho-mqtt 2.x で版を指定していない	mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
Warning: File … group is not mosquitto. Future versions will refuse to load this file.	パスワードファイルのグループが違う	chown mosquitto:mosquitto と chmod 600

同じクライアント ID で2つつないだときの、paho-mqtt の表示とブローカーのログです。

1つ目: 切断された reason_code=Unspecified error
2つ目: 切断された reason_code=Normal disconnection

New client connected from 127.0.0.1:44923 as device-001 (p2, c1, k60).
Client device-001 already connected, closing old connection.
New client connected from 127.0.0.1:57647 as device-001 (p2, c1, k60).

同じ ID の機器が2台あると、つなぐたびに相手を切り合い、自動で再接続する設定では切断と接続をくり返します。 ID には機器のシリアル番号や MAC アドレスを入れるのが一般的です。

マイコンから送るときのボード側のエラー（Wi-Fi・書き込み）は、関連記事「M5Stack・ESP32とは｜種類・開発環境・始め方とエラーの直し方」で扱っています。

よくある質問

Q. MQTT とは簡単に言うと何ですか？ A. ブローカーという中継サーバーを通して、トピックごとにメッセージを配る軽い通信プロトコルです。送る側は宛先を知らなくてよく、購読している全員に届きます。センサーや IoT の機器のデータを集めるのによく使われます。

Q. MQTT の読み方は？ A. エムキューティーティーと読みます。もとは Message Queuing Telemetry Transport の略ですが、今は略語ではなく正式な名前として扱われています。

Q. MQTT のポート番号は？ A. IANA に 1883 番（MQTT）と 8883 番（TLS を使う MQTT）が登録されています。WebSocket で使うポートは、ブローカーやサービスごとに決まります。

Q. MQTT のブローカーとは？無料で使えますか？ A. メッセージを受け取り、購読者に配る中継サーバーです。Mosquitto・HiveMQ Community Edition はオープンソースで、手元のマシンで無料で動かせます。クラウドのサービス（AWS IoT Core など）は有料で、料金は各サービスの料金ページに載っています。

Q. QoS 0・1・2 はどれを使えばよいですか？ A. 毎秒送る計測値のように1件欠けても次が来るものは QoS 0、欠けると困るが2回届いても受け側で扱えるものは QoS 1、2回処理すると困るものは QoS 2 が目安です。手元の測定では QoS 1 は QoS 0 の約4分の1、QoS 2 は約5分の1の速さでした。

Q. MQTT と HTTP の違いは？ A. HTTP は要求ごとに返事をもらう1対1の形、MQTT はつないだままブローカー経由で1対多に配る形です。2バイトの値を送ると、HTTP は要求と応答で173バイト、MQTT の QoS 0 は12バイトでした。

Q. Retain と永続セッションの違いは？ A. Retain はトピックごとに最後の1件だけを保存し、後から来た誰にでも渡します。永続セッションは、特定のクライアント ID のために、切断中のメッセージをすべてためておきます。

Q. Will（Last Will and Testament）とは？ A. 接続のときにブローカーへ預けておくメッセージで、正しく切断しなかったときにブローカーが代わりに発行します。DISCONNECT で正しく切ったときは発行されません。

Q. MQTT はブラウザから使えますか？ A. MQTT over WebSocket で使えます。ブローカーに WebSocket のリスナーを用意し、JavaScript のクライアントライブラリからつなぎます。

Q. MQTT 5 と 3.1.1 のどちらを使えばよいですか？ A. Mosquitto・EMQX・HiveMQ・AWS IoT Core はどちらにも対応しています。理由コード・メッセージの有効期限・共有サブスクリプションなどを使うなら MQTT 5 です。マイコンのライブラリが 3.1.1 だけに対応していることもあるので、機器側の対応を先に確かめます。

Q. MQTT と Kafka の違いは？ A. MQTT は多くの機器と軽くつなぐための通信プロトコル、Kafka はメッセージをディスクに保存して後から読み直せるイベント基盤です。機器から MQTT で集め、サーバー側で Kafka に流す組み合わせもあります。Kafka の使い方は、関連記事「Kafka入門｜KRaftでトピック作成から消費まで最短手順」で扱っています。

まとめ

• MQTT は、ブローカーを中継してトピックごとにメッセージを配る、軽い発行・購読型のプロトコル。ポートは 1883（平文）と 8883（TLS）。
• トピックは / 区切り。購読では +（1階層）と #（残り全部。親自身も含む）が使え、# でも $SYS は届かない。
• QoS 0 は PUBLISH 1つ、QoS 1 は PUBACK まで2つ、QoS 2 は4つのパケット。速さは QoS 1 で約4分の1、QoS 2 で約5分の1。QoS 1 でもブローカーのキューの上限を超えると捨てられる。
• Retain は最後の1件を保存し、Will は異常切断で発行され、Keep Alive の1.5倍を過ぎると切断される。永続セッションは同じ ID・QoS 1 以上で切断中のメッセージを受け取れる。
• HTTP の173バイトに対し、MQTT は12バイトで同じ値を送れた。ブラウザからは MQTT over WebSocket。
• MQTT 5 では理由コード・有効期限・トピックエイリアス・共有サブスクリプションなどが加わった。

TCP の接続とパケットの読み方は、関連記事「ソケットとは｜TCPとUDPの違いをPythonの通信とtcpdumpで解説」で、tcpdump のフィルターの書き方は、関連記事「tcpdumpの使い方｜フィルタ式一覧と出力の読み方」で扱っています。ブラウザとサーバーの WebSocket は、関連記事「Node.js WebSocket実装｜ws・再接続・認証・nginxまで」で、センサーをつなぐボードの始め方は、関連記事「ラズパイ（Raspberry Pi）とは｜始め方・モデル比較・Lチカを実例で解説」で扱っています。

参考・出典

確認日はいずれも 2026年10月23日です。

• MQTT.org「MQTT: The Standard for IoT Messaging」：https://mqtt.org/
• MQTT.org「MQTT Specification」：https://mqtt.org/mqtt-specification/
• OASIS「MQTT Version 5.0」：https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html
• OASIS「MQTT Version 3.1.1」：http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html
• ISO「ISO/IEC 20922:2016 Message Queuing Telemetry Transport (MQTT) v3.1.1」：https://www.iso.org/standard/69466.html
• IANA「Service Name and Transport Protocol Port Number Registry」：https://www.iana.org/assignments/service-names-port-numbers/service-names-port-numbers.xhtml?search=mqtt
• AWS「MQTT とは何ですか?」：https://aws.amazon.com/jp/what-is/mqtt/
• AWS IoT Core Developer Guide「MQTT」：https://docs.aws.amazon.com/iot/latest/developerguide/mqtt.html
• Eclipse Mosquitto：https://mosquitto.org/
• Eclipse Mosquitto「Download」：https://mosquitto.org/download/
• Eclipse Mosquitto「mqtt(7)」：https://mosquitto.org/man/mqtt-7.html
• Eclipse Mosquitto「mosquitto.conf(5)」：https://mosquitto.org/man/mosquitto-conf-5.html
• Eclipse Mosquitto「Authentication methods」：https://mosquitto.org/documentation/authentication-methods/
• PyPI「paho-mqtt」：https://pypi.org/project/paho-mqtt/
• Eclipse Paho「Migrations（paho-mqtt 2.0）」：https://eclipse.dev/paho/files/paho.mqtt.python/html/migrations.html
• EMQX「Introduction to MQTT 5.0」：https://www.emqx.com/en/blog/introduction-to-mqtt-5
• GitHub「emqx/emqx」：https://github.com/emqx/emqx
• GitHub「hivemq/hivemq-community-edition」：https://github.com/hivemq/hivemq-community-edition
• HiveMQ「MQTT Topics, Wildcards, & Best Practices」：https://www.hivemq.com/blog/mqtt-essentials-part-5-mqtt-topics-best-practices/
• HiveMQ「MQTT 5 Essentials Part 2: Foundational Changes in the Protocol」：https://www.hivemq.com/blog/mqtt5-essentials-part2-foundational-changes-in-the-protocol/

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。