はじめに
個人で開発している IoT のプロジェクトで Raspberry Pi と複数の Raspberry Pi Pico 2 W を通信させています。
もともとは HTTP で通信していましたが、実際に IoT デバイスを動かしていく中で、
- デバイスの生存確認をしたい
- 通信に失敗した場合、適切なタイミングでデータを再送したい
- 電源断などが起こった場合、再起動後に必要なデータを再送したい
- デバイスの IP アドレスに依存せず、サーバーからデバイスへ命令を送りたい
みたいな仕組みが欲しくなりました。
これらを実現するための仕組みを HTTP の上に自前で実装していましたが、調べてみたところ、やりたいことの多くが MQTT の機能として用意されていました。
この記事では、HTTP ではどのように実装していたのか、それを MQTT ではどのように実現できるのかを比較しながら、MQTT の仕組みを紹介します。
MQTT とは
MQTT は Pub/Sub 型の軽量なメッセージングプロトコルです。
仕様だと、IoT のようにコードサイズやネットワーク帯域が限られる環境に向いていると説明されています。
TCP の上で動き、一度張った接続を維持したままメッセージをやり取りします。
HTTP のリクエスト / レスポンスとは違い、送る側と受け取る側は直接やり取りせず、間に Broker を挟みます。
登場するのは Publisher、Subscriber、Broker の 3 つと、メッセージの宛先になる Topic です。
仕様は OASIS で標準化されています。
Publisher / Subscriber
MQTT では、メッセージを送る側を Publisher、受け取る側を Subscriber と呼びます。
HTTP のように相手を直接指定して送るのではなく、Publisher は Broker に向けて送り、Subscriber は Broker から受け取ります。
1 つのクライアントが Publisher, Subscriber の両方を兼ねることもできます。
Broker
Broker はメッセージの中継役です。
Publisher から受け取ったメッセージを、その topic を購読している Subscriber に配ります。
HTTP のときは、Raspberry Pi と Pico 2 W が相手の IP アドレスを指定して直接通信していましたが、MQTT では、どちらも Broker に接続してメッセージをやり取りするため、お互いの IP アドレスを知る必要がありません。
Pico 2 W <-> Broker <-> Raspberry Pi
Pico 2 W と Raspberry Pi はどちらも Publisher, Subscriber になれるため、双方向にメッセージをやり取りすることができます。
Topic
Topic は MQTT においてメッセージを分類するための名前です。
Publisher は Topic を指定してメッセージを送り、Subscriber は受け取りたい Topic を購読します。
例えば、デバイスごとに Topic を分ける場合は次のようにできます。
app/device/{device_id}/status
app/device/{device_id}/command
device_id にはアプリケーション側で割り当てた ID や、MAC アドレスなど、デバイスを一意に識別できる値を使えます。
/ で階層を分けることができるので、デバイスの ID やメッセージの種類ごとに Topic を整理できます。
Subscriber 側ではワイルドカードを使って複数の Topic をまとめて購読することもできます。
app/device/+/status
app/device/#
+ は 1 階層にマッチするため、すべてのデバイスの status をまとめて購読できます。
# はそれ以下のすべての階層にマッチするため、app/device/# で各デバイスの status や command などをまとめて購読できます。
HTTP で実装していたことを MQTT で実現する
デバイスの生存確認
私がしていた HTTP の実装では、Pico 2 W が 5 秒ごとに heartbeat を送り、一定時間届かなければサーバー（Raspberry Pi）側で offline とみなしていました。
MQTT では LWT（Last Will and Testament） を使うことができます。
接続時に「自分が落ちたらこれを publish してほしい」というメッセージを Broker に預けておくと、接続が切れたときに Broker が代わりに publish してくれます。
接続が切れたことは Keep Alive などによって検知されるため、アプリケーション側で heartbeat のタイムアウトを管理する必要がなくなります。
サーバーは status の Topic を購読するだけで済みます。
※ LWT（Last Will and Testament） と呼ばれることが多いですが、MQTT の公式仕様では Will Message と表記されています。
通信に失敗したデータの再送
HTTP では、送信したいデータを送る前にファイルへ追記しておき、届いたら消し、届かなかった分はあとでまとめて送り直す、みたいなことをしていました。
MQTT では、メッセージごとに QoS（Quality of Service）を指定して、どのような配送保証をするかを選べます。
| QoS | 意味 |
|---|---|
| 0 | 最大 1 回（送りっぱなし） |
| 1 | 少なくとも 1 回（重複はありうる） |
| 2 | ちょうど 1 回 |
QoS1 では、送る側が PUBLISH パケットを送ると、受け取った側が PUBACK パケットを返します。
送る側は PUBACK が返ってきたことで、メッセージが届いたと分かります。
QoS2 では PUBLISH → PUBREC → PUBREL → PUBCOMP というやり取りを行うことで、メッセージが重複して配送されることを防ぎます。その分、QoS を上げるほど通信のやり取りも増えます。
QoS は Publisher と Broker の間、Broker と Subscriber の間でそれぞれ適用されます。
※ ただし、QoS はデバイス上の永続的な再送キューそのものではないです。デバイスの電源断や再起動後にも未送信データを残したい場合は、必要に応じてファイルなどへの永続化も併用する必要があります。
再起動後に必要なデータの再送
サーバー（Raspberry Pi）から Pico 2 W に送った命令は、Pico 2 W のメモリにしか残らないので、電源断などで再起動すると消えてしまいます。
HTTP では、Pico 2 W が起動したことをサーバー側で記録しておき、heartbeat が届くたびに、まだ送っていなければ命令を送り直していました。
MQTT では Retained Message が使えます。
retain を付けて publish すると、Broker が Topic ごとに最新の 1 件を持っておき、あとから購読したクライアントにも渡してくれます。
Pico 2 W は再起動して購読し直すだけで最新の命令を受け取れるので、サーバー側で送り直す必要がなくなりました。
IP アドレスに依存しない命令の送信
私の HTTP を使った実装では、Raspberry Pi から Pico 2 W に命令を送るために、Pico 2 W 側でも HTTP サーバーを動かし、デバイス ID と IP アドレスの対応を Raspberry Pi 側で管理していました。
MQTT では、app/device/{device_id}/command のように Topic にデバイス ID を入れておけば、それが宛先になります。
Pico 2 W は自分の Topic を購読しておくだけでよく、HTTP サーバーを動かす必要もなくなります。
実装
今回は Raspberry Pi 上で Mosquitto を MQTT Broker として動かし、Raspberry Pi 上のアプリケーションは paho-mqtt、Pico 2 W は umqtt.simple を使って Broker に接続します。
Raspberry Pi
Broker（Mosquitto）
まず Raspberry Pi 上で MQTT Broker として Mosquitto を動かします。
sudo apt install mosquitto
Pico 2 W から接続できるように設定を追加します。
Mosquitto 2.0 以降は、listener を設定しないと localhost からの接続しか受け付けません。
今回はインターネットには接続しない閉じたネットワーク内で使用するため、認証は設定していません。
listener 1883
allow_anonymous true
persistence true
persistence true にしておくと、Mosquitto が保持している Retained Message などを永続化できます。
設定を変更したら Mosquitto を再起動します。
sudo systemctl restart mosquitto
Publisher / Subscriber（paho-mqtt）
Raspberry Pi 上で動くアプリケーションも MQTT Client の一つです。
Python では paho-mqtt を使って Mosquitto に接続し、Pico 2 W から送られてくるデータを subscribe したり、Pico 2 W への命令を publish したりします。
import paho.mqtt.client as mqtt
def on_connect(client, userdata, flags, reason_code, properties):
client.subscribe("app/device/+/#", qos=1)
def on_message(client, userdata, message):
print(message.topic, message.payload)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect("127.0.0.1", 1883)
client.loop_forever()
on_connect の中で subscribe しておくことで、Broker への再接続後にも必要な Topic を購読し直せます。
Pico 2 W に再起動後も受け取ってほしい命令は、retain=True を指定して publish します。
import json
client.publish(
"app/device/pico-01/command",
json.dumps(command),
qos=1,
retain=True,
)
Raspberry Pi Pico 2 W
Pico 2 W も MQTT Client として Broker に接続します。
MicroPython では umqtt.simple を使うことができます。
LWT は CONNECT パケットの一部として Broker に渡されるため、connect() より前に設定します。
broker_host には Raspberry Pi の IP アドレスを指定します（Wi-Fi には接続済みの前提です）。
from umqtt.simple import MQTTClient
status_topic = b"app/device/pico-01/status"
client = MQTTClient(b"pico-01", broker_host, port=1883, keepalive=6)
client.set_last_will(status_topic, b'{"online": false}', retain=True, qos=1)
client.connect()
client.publish(status_topic, b'{"online": true}', retain=True, qos=1)
online / offline を Retained Message にしておくことで、Raspberry Pi 側があとから購読しても、デバイスの最新の状態が分かります。
センサーデータは publish し、Raspberry Pi からの命令は自分宛ての Topic を subscribe して受け取ります。
def on_message(topic, msg):
print(topic, msg)
client.set_callback(on_message)
client.subscribe(b"app/device/pico-01/command", qos=1)
client.publish(b"app/device/pico-01/telemetry/live", payload, qos=0)
umqtt.simple は、check_msg() をループの中で定期的に呼んでメッセージを受信する作りになっています。
また、Keep Alive の PINGREQ を定期的に送る仕組みもないので、ping() も自分で呼びます。
import time
while True:
client.ping()
client.check_msg()
time.sleep(2)