BLOG
2019.11.18
MQTTライブラリ Paho Python を理解しようとしてみる
こんにちは、R＆Mグループの内田です。
最近、IoT向けといわれているプロトコル「MQTT」を使い始めてみました。
様々なモノとモノがつながるIoTでは各デバイス・サーバの間でデータの受け渡しが発生します。
MQTTは軽量であること、ネットワーク環境が不安定でも動作すること、1対多通信などの特徴があり、IoT向けといわれているようです。
QoS(通信保障),Retain(最終メッセージ保存),will(遺言)など様々な機能が備わっています。
MQTTでは中継サーバである「Broker(ブローカ)」とクライアントである「Publisher(パブリッシャ)」、「Subscriber(サブスクライバ)」という3種類の役割が登場します。
主なメッセージ配信の流れは
- SubscriberはあらかじめどのTopic(アドレス)を購読するかbrokerに登録しておく
- PublisherがTopicを指定し、Brokerへメッセージを配信する
- Brokerはpublisherが指定したTopic購読中のSubscriber(複数可)へ配信する
といった形です。
- 参考: 絵で見てわかるIoT/センサの仕組みと活用<翔泳社>
- 著 : 株式会社ＮＴＴデータ,河村雅人,大塚紘史,小林佑輔,小山武士,宮崎智也,石黒佑樹,小島康平
さて、タイトルにあるPahoについてです。
※今回MQTTの各環境立ち上げについては割愛します。気になった方はネットで検索などしてみてください。
Pahoは、MQTTのクライアント(Publisher/Subscriber)を実装するためのライブラリです。
対応言語は複数あるようですがPythonを使いました。
- Eclipse Paho™ MQTT Python Client： https://github.com/eclipse/paho.mqtt.python
試しに以下のスクリプトを作ってみます。
Publisher：topic名="Topic1"に "test1"というメッセージを送信する
Subscriber ：topic名="Topic1"のメッセージを受け取る
Publisher(pub_test1.py)
# -*- coding: utf-8 -*-
import paho.mqtt.client as mqtt # ライブラリのimport
# MQTT Broker
MQTT_HOST = "MQTT_broker" # brokerのアドレス
MQTT_PORT = 1883 # brokerのport
MQTT_KEEP_ALIVE = 60 # keep alive
# broker接続時
def on_connect(mqttc, obj, flags, rc):
print("rc: " + str(rc)) # 接続結果表示
mqttc = mqtt.Client() #clientオブジェクト作成
mqttc.on_connect = on_connect # 接続時に実行するコールバック関数設定
mqttc.connect(MQTT_HOST, MQTT_PORT, MQTT_KEEP_ALIVE) # MQTT broker接続
mqttc.loop_start() # 処理開始
mqttc.publish("topic1", "test1") # topic名="Topic1"に "test1"というメッセージを送信
Subscriber(sub_test1.py)
import paho.mqtt.client as mqtt
# MQTT Broker
MQTT_HOST = "MQTT_broker" # brokerのアドレス
MQTT_PORT = 1883 # brokerのport
MQTT_KEEP_ALIVE = 60 # keep alive
# broker接続時
def on_connect(mqttc, obj, flags, rc):
print("rc: " + str(rc))
#メッセージ受信時
def on_message(mqttc, obj, msg):
print(msg.topic + " " + str(msg.qos) + " " + str(msg.payload))
mqttc = mqtt.Client()
mqttc.on_message = on_message # メッセージ受信時に実行するコールバック関数設定
mqttc.on_connect = on_connect
mqttc.connect(MQTT_HOST, MQTT_PORT, MQTT_KEEP_ALIVE)
mqttc.subscribe("topic1") # Topic名："topic1"を購読
mqttc.loop_forever() # 永久ループ
実行結果
# python pub_test1.py
rc: 0
# python sub_test1.py
rc: 0
topic1 0 b'test1'
無事メッセージを受け取ることができました。
次に、受け取ったメッセージ次第でメインのループの処理を変えたいなと思ったのですが
on_message側から、mainループに値を返す or 共有する方法はないのか調べてみました。
以下サイトを読み解いてみると
- Paho Python MQTT Client – Understanding Callbacks ： http://www.steves-internet-guide.com/mqtt-python-callbacks/
コールバック関数は別スレッドで動いて非同期なので
on_messageのコールバック関数で変数をglobal宣言をして値を共有するのが正解みたいです。
Subscriberを改造してみます。
Publisher：topic名="Topic1"に "test1"というメッセージを送信する
Subscriber ：topic名="Topic1"のメッセージを受け取ったらフラグ有効化し、ループを抜ける
Subscriber(sub_test2.py)
# -*- coding: utf-8 -*-
import paho.mqtt.client as mqtt
import time
# MQTT Broker
MQTT_HOST = "MQTT_broker" # brokerのアドレス
MQTT_PORT = 1883 # brokerのport
MQTT_KEEP_ALIVE = 60 # keep alive
# broker接続時
def on_connect(mqttc, obj, flags, rc):
print("rc: " + str(rc))
#メッセージ受信時
def on_message(mqttc, obj, msg):
global on_message_Flag # メインループと共有するグローバル変数
print(msg.topic + " " + str(msg.qos) + " " + str(msg.payload))
on_message_Flag=True # flagを有効化
mqttc = mqtt.Client()
mqttc.on_message = on_message
mqttc.on_connect = on_connect
mqttc.connect(MQTT_HOST, MQTT_PORT, MQTT_KEEP_ALIVE)
mqttc.subscribe("topic1")
mqttc.loop_start()
on_message_Flag=False # 共有するグローバル変数
i=0
while True:
print(i)
i+=1
time.sleep(1)
if on_message_Flag: # フラグがTrueになるとループEND
print("END!")
break
実行結果
# python pub_test1.py
rc: 0
# python sub_test2.py
0
rc: 0
1
2
3
4
5
topic1 0 b'test1'
END!
メッセージを受け取ったのち無事ループを抜けました。
いろいろインターネットで検索をかけたり、公式のexamplesを参考にすると、もっと色々なことができそうですね！
関連記事
一緒に開発しませんか？
サンビット株式会社では、開発技術者を募集しています！
興味のある方はぜひお問い合わせください。
