Python paho-mqtt 模塊使用（轉）

• 一、Client模塊
  • （一）使用流程
  • （二）回調(Callbacks)
    • 1.基本概念
    • 2.回調種類
      • （1）on_connect()
      • （2）on_disconnect()
      • （3）on_message()
      • （4）on_publish()
      • （5）on_subscribe()
      • （6）on_unsubscribe()
      • （7）on_log()
  • （三）方法
    • 1.構造函數Client（）
    • 2.reinitialise()
    • 3.選項函數
      • （1）max_inflight_messages_set()
      • （2）max_queued_messages_set()
      • （3）message_retry_set（）
      • （4）ws_set_options()
      • （4）tls_set()
      • （5）tls_set_context()
      • （6）tls_insecure_set()
      • （7）enable_logger()
      • （8）disable_logger()
      • （9）username_pw_set()
      • （10）user_data_set()
      • （11）will_set()
      • （11）reconnect_delay_set（）
    • 4.connect()
    • 5.connect_async()
    • 6.connect_srv()
    • 7.reconnect()
    • 8.disconnect()
    • 9.loop()
    • 10.loop_start() / loop_stop()
    • 11.loop_forever()
    • 12.publish()
    • 13.subscribe()
      • （1）簡單的字符串和整數
      • （2）字符串和整數元組
      • （3）字符串和整數元組的列表
    • 14.unsubscribe()
    • 15.外部事件循環支持
      • （1）loop_read()
      • （2）loop_write()
      • （3）loop_misc()
      • （4）socket()
      • （5）want_write()
    • 16.全局輔助函數
• 二、Publish模塊
  • （一）方法
    • 1.Single
    • 2.Multiple
• 三、Subscribe模塊
  • （1）方法
    • 1.Simple
    • 2.Callback

---

一、Client模塊

與MQTT代理（broker）進行通信的主要類。

---

（一）使用流程

• 使用connect()/connect_async() 連接MQTT代理

• 頻繁的調用loop()來維持與MQTT代理之間的流量

  • 或者使用loop_start()來設置一個線程為你調用loop()
  • 或者在一個阻塞的函數中調用loop_forever()來為你調用loop()

• 使用subscribe()訂閱一個主題（topic）並接受消息（messages）

• 使用publish()來發送消息

• 使用disconnect()來斷開與MQTT代理的連接

---

（二）回調(Callbacks)

1.基本概念

使用回調處理從MQTT代理返回的數據，要使用回調需要先定義回調函數然后將其指派給客戶端實例（client）。
例如：

# 定義一個回調函數 def on_connect(client, userdata, flags, rc): print("Connection returned " + str(rc)) # 將回調函數指派給客戶端實例 client.on_connect = on_connect 

所有的回調函數都有client和userdata參數。
client是調用回調的客戶端實例；
userdata可以使任何類型的用戶數據，可以在創建新客戶端實例時設置或者使用user_data_set(userdata)設置。

2.回調種類

（1）on_connect()

當代理響應連接請求時調用。
on_connect(client, userdata, flags, rc):
flags是一個包含代理回復的標志的字典；
rc的值決定了連接成功或者不成功：

值	連接情況
0	連接成功
1	協議版本錯誤
2	無效的客戶端標識
3	服務器無法使用
4	錯誤的用戶名或密碼
5	未經授權

（2）on_disconnect()

當與代理斷開連接時調用

on_disconnect(client, userdata, rc):

rc參數表示斷開狀態。
如果MQTT_ERR_SUCCESS（0），回調被調用以響應disconnect（）調用。 如果以任何其他值斷開連接是意外的，例如可能出現網絡錯誤。

（3）on_message()

on_message(client, userdata, message):

當收到關於客戶訂閱的主題的消息時調用。
message是一個描述所有消息參數的MQTTMessage。

（4）on_publish()

當使用使用publish()發送的消息已經傳輸到代理時被調用。

on_publish(client, userdata, mid):

對於Qos級別為1和2的消息，這意味着已經完成了與代理的握手。
對於Qos級別為0的消息，這只意味着消息離開了客戶端。
mid變量與從相應的publish()返回的mid變量匹配，以允許跟蹤傳出的消息。

此回調很重要，因為即使publish（）調用返回，但並不總意味着消息已發送。

（5）on_subscribe()

當代理響應訂閱請求時被調用。

on_subscribe(client, userdata, mid, granted_qos):

mid變量匹配從相應的subscri be()返回的mid變量。
‘granted_qos’變量是一個整數列表，它提供了代理為每個不同的訂閱請求授予的QoS級別。

（6）on_unsubscribe()

當代理響應取消訂閱請求時調用。

on_unsubscribe(client, userdata, mid):

mid匹配從相應的unsubscribe（）調用返回的中間變量。

（7）on_log()

當客戶端有日志信息時調用

on_log(client, userdata, level, buf):

level變量給出了消息的嚴重性，並且將是MQTT_LOG_INFO，MQTT_LOG_NOTICE，MQTT_LOG_WARNING，MQTT_LOG_ERR和MQTT_LOG_DEBUG中的一個。
buf變量用於存儲信息。

---

（三）方法

1.構造函數Client（）

Client(client_id="", clean_session=True, userdata=None, protocol=MQTTv311, transport="tcp") 

參數	含義
client_id	連接到代理時使用的唯一客戶端ID字符串。 如果client_id長度為零或無，則會隨機生成一個。 在這種情況下，clean_session參數必須為True。
clean_session	一個決定客戶端類型的布爾值。 如果為True，那么代理將在其斷開連接時刪除有關此客戶端的所有信息。 如果為False，則客戶端是持久客戶端，當客戶端斷開連接時，訂閱信息和排隊消息將被保留。
userdata	用戶定義的任何類型的數據作為userdata參數傳遞給回調函數。 它可能會在稍后使用user_data_set（）函數進行更新。
protocol	用於此客戶端的MQTT協議的版本。 可以是MQTTv31或MQTTv311。
transport	設置為“websockets”通過WebSockets發送MQTT。 保留默認的“tcp”使用原始TCP。

示例：

import paho.mqtt.client as mqtt client = mqtt.Client() 

2.reinitialise()

reinitialise(client_id="", clean_session=True, userdata=None) 

reinitialise（）函數將客戶端重置為其開始狀態，就像它剛剛創建一樣。 它采用與Client（）構造函數相同的參數。
示例：

client.reinitialise() 

3.選項函數

這些函數表示可以在客戶端上設置以修改其行為的選項。 在大多數情況下，這必須在連接到代理之前完成。

（1）max_inflight_messages_set()

max_inflight_messages_set(self, inflight) 

設置QoS> 0的消息的最大數量，該消息一次可以部分通過其網絡流量。默認為20.增加此值將消耗更多內存，但可以增加吞吐量。

（2）max_queued_messages_set()

max_queued_messages_set(self, queue_size) 

設置傳出消息隊列中可等待處理的具有QoS> 0的傳出消息的最大數量。默認為0表示無限制。 當隊列已滿時，任何其他傳出的消息都將被丟棄。

（3）message_retry_set（）

message_retry_set(retry) 

如果代理沒有響應，設置在重發QoS> 0的消息之前以秒為單位的時間。默認設置為5秒，通常不需要更改。

（4）ws_set_options()

ws_set_options(self, path="/mqtt", headers=None) 

設置websocket連接選項。 只有在transport =“websockets”被傳入Client（）構造函數時才會使用這些選項。

參數	含義
path	代理使用的mqtt路徑
headers	可以是一個字典，指定應該附加到標准websocket頭部的額外頭部列表，也可以是可調用的正常websocket頭部並返回帶有一組頭部以連接到代理的新字典。

必須在調用connect()之前調用。

（4）tls_set()

tls_set(ca_certs=None, certfile=None, keyfile=None, cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS, ciphers=None) 

配置網絡加密和身份驗證選項。 啟用SSL / TLS支持。

參數	含義
ca_certs	證書頒發機構證書文件的字符串路徑，該證書文件將被視為受此客戶端信任。
certfile, keyfile	分別指向PEM編碼的客戶端證書和私鑰的字符串。 如果這些參數不是None，那么它們將用作基於TLS的身份驗證的客戶端信息。 對此功能的支持取決於代理。
cert_reqs	定義客戶對經紀人施加的證書要求。 默認情況下，這是ssl.CERT_REQUIRED，這意味着代理必須提供證書。 有關此參數的更多信息，請參閱ssl pydoc。
tls_version	指定要使用的SSL / TLS協議的版本。 默認情況下（如果python版本支持它），檢測到最高的TLS版本。
ciphers	指定哪些加密密碼可供此連接使用的字符串，或者使用None來使用默認值。 有關更多信息，請參閱ssl pydoc。

必須在調用connect()之前調用。

（5）tls_set_context()

配置網絡加密和認證上下文。 啟用SSL / TLS支持。

tls_set_context(context=None) 

參數	含義
context	一個ssl.SSLContext對象。

必須在調用connect()之前調用。

（6）tls_insecure_set()

配置服務器證書中服務器主機名的驗證。

tls_insecure_set(value)

如果value設置為True，則不可能保證您連接的主機不模擬您的服務器。 這在初始服務器測試中可能很有用，但是，惡意的第三方通過可以DNS欺騙模擬您的服務器。

• 請勿在真實系統中使用此功能。 將值設置為True意味着使用加密沒有意義。
• 必須在connect*（）之前和tls_set（）或tls_set_context（）之后調用。

（7）enable_logger()

使用標准的Python日志包啟用日志記錄。 這可以與on_log回調方法同時使用

enable_logger(logger=None) 

如果指定了記錄器，那么將使用該logging.Logger對象，否則將自動創建一個。
按照以下映射將Paho日志記錄級別轉換為標准日志級別：

Paho	logging
MQTT_LOG_ERR	ligging.ERROR
MQTT_LOG_WARNING	logging.WARNING
MQTT_LOG_NOTICE	logging.INFO (no direct equivalent)
MQTT_LOG_INFO	logging.INFO
MQTT_LOG_DEBUG	logging.DEBUG

（8）disable_logger()

使用標准python日志包禁用日志記錄。 這對on_log回調沒有影響。

disable_logger()

（9）username_pw_set()

為代理認證設置一個用戶名和一個可選的密碼。必須在connect*（）之前調用。

username_pw_set(username, password=None) 

（10）user_data_set()

設置在生成事件時將傳遞給回調的私人用戶數據。 將其用於您自己的目的以支持您的應用程序。

user_data_set(userdata)

（11）will_set()

設置要發送給代理的遺囑。 如果客戶端斷開而沒有調用disconnect（），代理將代表它發布消息。

will_set(topic, payload=None, qos=0, retain=False) 

參數	含義
topic	該遺囑消息發布的主題
payload	該消息將作為遺囑發送
qos	用於遺囑的服務質量等級
retain	如果設置為True，遺囑消息將被設置為該主題的“最后已知良好”/保留消息。

如果qos不是0,1或2，或者主題為None或字符串長度為零，則引發ValueError。

（11）reconnect_delay_set（）

客戶端將自動重試連接。 在每次嘗試之間，它會在min_delay和max_delay之間等待幾秒鍾。

reconnect_delay_set(min_delay=1, max_delay=120)

當連接丟失時，最初重新連接嘗試延遲min_delay秒。 延遲在隨后的嘗試到中增加一倍。當連接完成時（例如收到CONNACK，而不僅僅是TCP連接建立），延遲重置為min_delay。

4.connect()

connect（）函數將客戶端連接到代理。 這是一個阻塞函數。

connect(host, port=1883, keepalive=60, bind_address="") 

參數	含義
host	遠程代理的主機名或IP地址
port	要連接的服務器主機的網絡端口。 默認為1883
keepalive	與代理通信之間允許的最長時間段（以秒為單位）。 如果沒有其他消息正在交換，則它將控制客戶端向代理發送ping消息的速率
bind_address	假設存在多個接口，將綁定此客戶端的本地網絡接口的IP地址

5.connect_async()

與loop_start（）一起使用以非阻塞方式連接。 直到調用loop_start（）之前，連接才會完成。

connect_async(host, port=1883, keepalive=60, bind_address="") 

6.connect_srv()

使用SRV DNS查找連接到代理以獲取代理地址。

connect_srv(domain, keepalive=60, bind_address="") 

參數	含義
domain	該DNS域搜索SRV記錄。 如果無，請嘗試確定本地域名。

7.reconnect()

使用先前提供的詳細信息重新連接到經紀商。 在調用此函數之前，您必須先調用connect *（）。

reconnect()

8.disconnect()

干凈地從代理斷開連接。 使用disconnect（）不會導致代理發送遺囑消息。

disconnect()

9.loop()

定期調用處理網絡事件。

loop(timeout=1.0, max_packets=1)

此調用在select（）中等待，直到網絡套接字可用於讀取或寫入（如果適用），然后處理傳入/傳出數據。

參數	含義
timeout	此方法最多可阻塞timeout秒
max_packets	max_packets參數已過時，應保留為未設置狀態。

示例：

run = True while run: client.loop() 

10.loop_start() / loop_stop()

這些功能實現了到網絡循環的線程接口。

loop_start()
loop_stop(force=False) 

在connect*（）之前或之后調用loop_start（）一次，會在后台運行一個線程來自動調用loop（）。這釋放了可能阻塞的其他工作的主線程。這個調用也處理重新連接到代理。
調用loop_stop（）來停止后台線程。

force參數目前被忽略。
示例：

client.connect("iot.eclipse.org") client.loop_start() while True: temperature = sensor.blocking_read() client.publish("paho/temperature", temperature) 

11.loop_forever()

這是網絡循環的阻塞形式，直到客戶端調用disconnect（）時才會返回。它會自動處理重新連接。

loop_forever(timeout=1.0, max_packets=1, retry_first_connection=False) 

除了使用connect_async時的第一次連接嘗試以外，請使用retry_first_connection = True使其重試第一個連接。這可能會導致客戶端連接到一個不存在的主機的情況。

12.publish()

從客戶端發送消息給代理。

publish(topic, payload=None, qos=0, retain=False) 

消息將會發送給代理，並隨后從代理發送到訂閱匹配主題的任何客戶端。

參數	含義
topic	該消息發布的主題
payload	要發送的實際消息。如果沒有給出，或設置為無，則將使用零長度消息。 傳遞int或float將導致有效負載轉換為表示該數字的字符串。 如果你想發送一個真正的int / float，使用struct.pack（）來創建你需要的負載
qos	服務的質量級別
retain	如果設置為True，則該消息將被設置為該主題的“最后已知良好”/保留的消息

返回以下屬性和方法的MQTTMessageInfo:
rc:發布的結果。

內容	含義
MQTT_ERR_SUCCESS	成功
MQTT_ERR_NO_CONN	客戶端當前未連接
MQTT_ERR_QUEUE_SIZE	當使用max_queued_messages_set來指示消息既不排隊也不發送。

mid:發布請求的消息ID。
如果mid已定義，則可以通過檢查on_publish（）回調中的mid來跟蹤發布請求。
wait_for_publish():函數將阻塞，直到消息發布。 如果消息未排隊（rc == MQTT_ERR_QUEUE_SIZE），它將引發ValueError。
is_published:如果消息已發布，is_published返回True。 如果消息未排隊（rc == MQTT_ERR_QUEUE_SIZE），它將引發ValueError。

如果主題為無，長度為零或無效（包含通配符），qos不是0,1或2之一，或者有效負載長度大於268435455字節，則會引發ValueError。

13.subscribe()

subscribe(topic, qos=0)

訂閱一個或多個主題。
這個函數可以用三種不同的方式調用：

（1）簡單的字符串和整數

subscribe("my/topic", 2) 

參數	值
topic	一個字符串，指定要訂閱的訂閱主題
qos	期望的服務質量等級。 默認為0。

（2）字符串和整數元組

subscribe(("my/topic", 1)) 

參數	值
topic	（topic，qos）的元組。 主題和qos都必須存在於元組中。
qos	沒有使用

（3）字符串和整數元組的列表

這允許在單個SUBSCRIPTION命令中使用多個主題訂閱，這比使用多個訂閱subscribe（）更有效。

subscribe([("my/topic", 0), ("another/topic", 2)]) 

參數	值
topic	格式元組列表（topic，qos）。 topic和qos都必須出現在所有的元組中。
qos	沒有使用

該函數返回一個元組（result，mid）。

如果qos不是0,1或2，或者主題為None或字符串長度為零，或者topic不是字符串，元組或列表，則引發ValueError。

14.unsubscribe()

取消訂閱一個或多個主題。

unsubscribe(topic)

參數	含義
topic	主題的單個字符串或者字符串列表

返回一個元組(result, mid)

15.外部事件循環支持

（1）loop_read()

loop_read(max_packets=1)

當套接字准備好讀取時調用。 max_packets已過時，應保持未設置狀態。

（2）loop_write()

loop_write(max_packets=1)

當套接字准備好寫入時調用。 max_packets已過時，應保持未設置狀態。

（3）loop_misc()

loop_misc()

每隔幾秒呼叫一次以處理消息重試和ping。

（4）socket()

socket()

返回客戶端中使用的套接字對象，以允許與其他事件循環進行交互。

（5）want_write()

want_write()

如果有數據等待寫入，則返回true，以允許將客戶端與其他事件循環連接。

16.全局輔助函數

client模塊還提供了一些全局幫助函數。
(1)topic_matches_sub（sub，topic）可用於檢查主題是否與預訂匹配。
(2)connack_string（connack_code）返回與CONNACK結果關聯的錯誤字符串。
(3)error_string（mqtt_errno）返回與Paho MQTT錯誤號關聯的錯誤字符串。

---

二、Publish模塊

該模塊提供了一些幫助功能，可以以一次性方式直接發布消息。換句話說，它們對於您想要發布給代理的單個/多個消息然后斷開與其他任何必需的連接的情況非常有用。

（一）方法

1.Single

將一條消息發布給代理，然后徹底斷開連接。

single(topic, payload=None, qos=0, retain=False, hostname="localhost", port=1883, client_id="", keepalive=60, will=None, auth=None, tls=None, protocol=mqtt.MQTTv311, transport="tcp") 

參數	含義
topic	唯一必需的參數必須是負載將發布到的主題字符串。
payload	要發布的有效載荷。 如果“”或None，零長度的有效載荷將被發布
qos	發布時使用的qos默認為0
retain	設置消息保留（True）或不（False）
hostname	一個包含要連接的代理地址的字符串。 默認為localhost
port	要連接到代理的端口。 默認為1883
client_id	要使用的MQTT客戶端ID。 如果“”或None，Paho庫會自動生成客戶端ID
keepalive	客戶端的存活超時值。 默認為60秒
will	一個包含客戶端遺囑參數的字典,will = {‘topic’: “<topic>”, ‘payload’:”<payload”>, ‘qos’:<qos>, ‘retain’:<retain>}.
auth	一個包含客戶端驗證參數的字典,auth = {‘username’:”<username>”, ‘password’:”<password>”}
tls	一個包含客戶端的TLS配置參數的字典,dict = {‘ca_certs’:”<ca_certs>”, ‘certfile’:”<certfile>”, ‘keyfile’:”<keyfile>”, ‘tls_version’:”<tls_version>”, ‘ciphers’:”<ciphers”>}
protocol	選擇要使用的MQTT協議的版本。 使用MQTTv31或MQTTv311。
transport	設置為“websockets”通過WebSockets發送MQTT。 保留默認的“tcp”使用原始TCP。

示例：

import paho.mqtt.publish as publish publish.single("paho/test/single", "payload", hostname="iot.eclipse.org") 

2.Multiple

將多條消息發布給代理，然后干凈地斷開連接。

multiple(msgs, hostname="localhost", port=1883, client_id="", keepalive=60, will=None, auth=None, tls=None, protocol=mqtt.MQTTv311, transport="tcp") 

參數	含義
msgs	要發布的消息列表。 每條消息是一個字典或一個元組。msg = {‘topic’:”<topic>”, ‘payload’:”<payload>”, ‘qos’:<qos>, ‘retain’:<retain>}或(“<topic>”, “<payload>”, qos, retain)

有關hostname，port，client_id，keepalive，will，auth，tls，protocol，transport的描述，請參閱single（）。
示例：

import paho.mqtt.publish as publish msgs = [{'topic':"paho/test/multiple", 'payload':"multiple 1"}, ("paho/test/multiple", "multiple 2", 0, False)] publish.multiple(msgs, hostname="iot.eclipse.org") 

---

三、Subscribe模塊

該模塊提供了一些幫助功能，以允許直接訂閱和處理消息。

（1）方法

1.Simple

訂閱一組主題並返回收到的消息。 這是一個阻塞函數。

simple(topics, qos=0, msg_count=1, retained=False, hostname="localhost", port=1883, client_id="", keepalive=60, will=None, auth=None, tls=None, protocol=mqtt.MQTTv311) 

參數	含義
topics	唯一需要的參數是客戶端將訂閱的主題字符串。 如果需要訂閱多個主題，這可以是字符串或字符串列表
qos	訂閱時使用的qos默認為0
msg_count	從代理檢索的消息數量。 默認為1.如果為1，則返回一個MQTTMessage對象。 如果> 1，則返回MQTTMessages列表
retained	設置為True以考慮保留的消息，將其設置為False以忽略具有保留標志設置的消息
hostname	一個包含要連接的代理地址的字符串。 默認為localhost
port	要連接到代理的端口。 默認為1883
client_id	要使用的MQTT客戶端ID。 如果“”或None，Paho庫會自動生成客戶端ID
keepalive	客戶端的存活超時值。 默認為60秒。
will	一個包含客戶端遺囑參數的字典,will = {‘topic’: “<topic>”, ‘payload’:”<payload”>, ‘qos’:<qos>, ‘retain’:<retain>}.
auth	一個包含客戶端驗證參數的字典,auth = {‘username’:”<username>”, ‘password’:”<password>”}
tls	一個包含客戶端的TLS配置參數的字典,dict = {‘ca_certs’:”<ca_certs>”, ‘certfile’:”<certfile>”, ‘keyfile’:”<keyfile>”, ‘tls_version’:”<tls_version>”, ‘ciphers’:”<ciphers”>}
protocol	選擇要使用的MQTT協議的版本。 使用MQTTv31或MQTTv311。

2.Callback

訂閱一組主題並使用用戶提供的回叫處理收到的消息。

callback(callback, topics, qos=0, userdata=None, hostname="localhost", port=1883, client_id="", keepalive=60, will=None, auth=None, tls=None, protocol=mqtt.MQTTv311) 

參數	含義
callback	一個“on_message”回調將被用於每個收到的消息
topics	客戶端將訂閱的主題字符串。 如果需要訂閱多個主題，這可以是字符串或字符串列表。
qos	訂閱時使用的qos默認為0
userdata	用戶提供的對象將在收到消息時傳遞給on_message回調函數

有關hostname，port，client_id，keepalive，will，auth，tls，protocol的描述，請參閱simple（）。
示例：

import paho.mqtt.subscribe as subscribe def on_message_print(client, userdata, message): print("%s %s" % (message.topic, message.payload)) subscribe.callback(on_message_print, "paho/test/callback", hostname="iot.eclip

參考資料： https://pypi.python.org/pypi/paho-mqtt

作者：whele
鏈接：https://www.jianshu.com/p/ef546f476322
來源：簡書
簡書著作權歸作者所有，任何形式的轉載都請聯系作者獲得授權並注明出處。