Python クライアント

準備

pahoのインストール

pip install paho-mqtt

プロジェクトディレクトリの作成

mkdir python-mqtt && cd python-mqtt

パブリッシャー

クライアント

import paho.mqtt.client as mqtt

mqttClient = mqtt.Client("python_pub") # パブリッシャー名

接続コールバック

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("CONNACK OK")
    else:
        print("CONNACK KO code=", rc)

接続（TLSなし）

MQTTの平文ソケットでmachbase-neoに接続します。

mqttClient = mqtt.Client("python_pub", clean_session=True)
mqttClient.on_connect = on_connect
mqttClient.connect("127.0.0.1", port=5653, keepalive=10, clean_session=True)
mqttClient.loop_start()

切断

mqttClient.disconnect()
mqttClient.loop_stop()

発行コールバック

def on_publish(client, userdata, mid):
    print("PUBACK mid=",mid)

発行

mqttClient.on_publish = on_publish

mqttClient.publish("db/append/example", """[
    ["temperature",1677033057000000000, 21.1],
    ["humidity",   1677033057000000000, 0.53]
]""", qos=1)

ソースコード全体

import paho.mqtt.client as mqtt
import time

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("CONNACK OK")
    else:
        print("CONNACK KO code:", rc)

def on_publish(client, userdata, mid):
    print("PUBACK mid:",mid)

mqttClient = mqtt.Client("python_pub", clean_session=True)
mqttClient.on_connect = on_connect
mqttClient.on_publish = on_publish
mqttClient.connect("127.0.0.1", port=5653, keepalive=10)
mqttClient.loop_start()

mqttClient.publish("db/append/example", """[
    ["temperature",1677033057000000000, 21.1],
    ["humidity",   1677033057000000000, 0.53]
]""", qos=1)

time.sleep(1)

mqttClient.disconnect()
mqttClient.loop_stop()

最終更新日 9月 10, 2026

C# クライアント Go クライアント