BLOG

2019.11.18

MQTTライブラリ Paho Python を理解しようとしてみる

このエントリーをはてなブックマークに追加

 こんにちは、R&Mグループの内田です。

最近、IoT向けといわれているプロトコル「MQTT」を使い始めてみました。

様々なモノとモノがつながるIoTでは各デバイス・サーバの間でデータの受け渡しが発生します。

MQTTは軽量であること、ネットワーク環境が不安定でも動作すること、1対多通信などの特徴があり、IoT向けといわれているようです。
QoS(通信保障),Retain(最終メッセージ保存),will(遺言)など様々な機能が備わっています。

MQTTでは中継サーバである「Broker(ブローカ)」とクライアントである「Publisher(パブリッシャ)」、「Subscriber(サブスクライバ)」という3種類の役割が登場します。

主なメッセージ配信の流れは

  1. SubscriberはあらかじめどのTopic(アドレス)を購読するかbrokerに登録しておく
  2. PublisherがTopicを指定し、Brokerへメッセージを配信する
  3. Brokerはpublisherが指定したTopic購読中のSubscriber(複数可)へ配信する

といった形です。

  • 参考: 絵で見てわかるIoT/センサの仕組みと活用<翔泳社> 
  • 著    : 株式会社NTTデータ,河村雅人,大塚紘史,小林佑輔,小山武士,宮崎智也,石黒佑樹,小島康平

さて、タイトルにあるPahoについてです。
※今回MQTTの各環境立ち上げについては割愛します。気になった方はネットで検索などしてみてください。

Pahoは、MQTTのクライアント(Publisher/Subscriber)を実装するためのライブラリです。
対応言語は複数あるようですが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 共有する方法はないのか調べてみました。

以下サイトを読み解いてみると

コールバック関数は別スレッドで動いて非同期なので
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を参考にすると、もっと色々なことができそうですね!

カレンダー

«11月»
     1 2
3 4 5 6 7 8 9
10 11 12 13 14 15 16
17 18 19 20 21 22 23
24 25 26 27 28 29 30

ブログ内検索