每天一点小进步(7):Mqtt客户端理解

网上找了很多关于Mqtt 客户端变成的文章，基本是翻译官方文档为主，从小白角度来看，这些文档还不足以让一个小白能够快速上手并完成简单客户端的开发。

于是自己总结了一些内容，用于练习并熟练记忆。

第一步：安装mqtt库

pip install paho-mqtt

第二步：创建mqtt客户端脚本

创建一个test_mqtt_client.py文件

第三步：在文件中编写脚本

1、简单的客户端脚本-Mqtt服务器需要做任何校验

import paho.mqtt.client as mqtt

MQTTHOST = "101.200.46.138"
MQTTPORT = 1883
mqttClient = mqtt.Client()
# 连接MQTT服务器
def on_mqtt_connect():
  mqttClient.connect(MQTTHOST, MQTTPORT, 60)
  mqttClient.loop_start()
# publish 消息
def on_publish(topic, payload, qos):
  mqttClient.publish(topic, payload, qos)
# 消息处理函数
def on_message_come(lient, userdata, msg):
  print(msg.topic + " " + ":" + str(msg.payload))
# subscribe 消息
def on_subscribe():
  mqttClient.subscribe("/server", 1)
  mqttClient.on_message = on_message_come # 消息到来处理函数
def main():
  on_mqtt_connect()
  on_publish("/test/server", "Hello Python!", 1)
  on_subscribe()
  while True:
    pass
if __name__ == '__main__':
  main()

2、一般项目中使用的Mqtt服务都需要认证，以下是认证客户端脚本

# -*- coding:utf-8 -*-

# /usr/bin/env python 

__author__ = 'Five'

import json
import time

import paho.mqtt.client as mqtt

from models.model_wx_pc import PCShow

pc = PCShow()

# 鉴权服务获取秘钥
def getuserTicker(uid):
	content = pc.userTicker(uid)
	return content["data"]["ext"], content["data"]["ticker"]


def on_connect(mqttc, obj, flags, rc):
	print("Connected with result code " + str(rc) + ' '+str(flags) + obj)


def on_message(mqttc, obj, msg):
	print("message: "+ msg)


def on_subscribe(mqttc, obj, mid, granted_qos):
	print(obj + "订阅Subscribed: " + str(mid) + " " + str(granted_qos))


def on_log(mqttc, obj, level, string):
	print(obj +level+ u'日志' + string)


def on_publish(mqttc, obj, mid):
	print(obj + u"发送消息" + str(mid))


def client(host='101.200.46.138', uid="1"):
	ext, ticker = getuserTicker(uid)
	topic = 'default'
	port = 1883
	payload = {
		'cmd':'inroom',
		'msgid':1
}
        # 初始化mqtt实例时需要根据mqtt服务端具体协议版本指定protocol
        # MQTTv31 = 3，MQTTv311 = 4，MQTTv5 = 5一定要匹配，不然无法连接服务器，其他参数详解看源码即可
	mqttc = mqtt.Client(client_id="1", clean_session=True, protocol=3)
       # 设置mqtt实例的认证信息
	mqttc.username_pw_set(ticker, ticker)
       # 设置mqtt实例中自定义的消息内容
	mqttc.user_data_set(u'无敌是多么寂寞')
	mqttc.on_message = on_message
	mqttc.on_connect = on_connect
	mqttc.on_subscribe = on_subscribe
	mqttc.on_publish = on_publish
	mqttc.on_log=on_log
	mqttc.connect(host, port, 60)
        # 发布
	mqttc.publish(topic, payload, 0)
        # 订阅
	mqttc.subscribe(topic, 0)
        # loop_forever调用会阻塞主线程，将永远不会终止
	mqttc.loop_forever()
			

if __name__ == '__main__':
	client()

3、第2种方式用使用了loop_forever，导致主线程一直阻塞无法完成其他的事情，优化脚本使用loop_start()（Loop_start在另一个线程中启动一个循环，如果您需要在主线程中执行其他操作，则让主线程继续运行其他内容。），并在主线中打印消息

# -*- coding:utf-8 -*-

# /usr/bin/env python 

__author__ = 'Five'

import json
import time

import paho.mqtt.client as mqtt

from models.model_wx_pc import PCShow

from queue import Queue

q = Queue()
msgs = []

pc = PCShow(1083722)


def getuserTicker(uid=607538990):
	content = PCShow().userTicker(uid)
	return content["data"]["ext"], content["data"]["ticker"]


def on_connect(mqttc, obj, flags, rc):
	print("Connected with result code " + str(rc) + ' '+str(flags) + obj)


def on_message(mqttc, obj, msg):
	# if json.loads(msg.payload.decode('utf-8'))["cmd"] == "onpwsysgiftbox":
	# 	print(json.loads(msg.payload.decode('utf-8'))['activity_ret']['data'])
	m = "message received  ", str(msg.payload.decode("utf-8"))
	msgs.append(m)
	q.put(m)


def on_subscribe(mqttc, obj, mid, granted_qos):
	print(obj + "订阅Subscribed: " + str(mid) + " " + str(granted_qos))


def on_log(mqttc, obj, level, string):
	print(obj +level+ u'日志' + string)


def on_publish(mqttc, obj, mid):
	print(obj + u"发送消息" + str(mid))


def client(host='101.200.46.138', uid="1"):
	ext, ticker = getuserTicker(uid)
	topic = 'default'
	port = 1883
	payload = {
		"Cmd": "inroom",
		"Platform": "Android",
		"MsgId": 99,
	}
	mqttc = mqtt.Client(client_id="1", clean_session=True, protocol=3)
	mqttc.username_pw_set(ticker, ticker)
	mqttc.user_data_set(u'无敌是多么寂寞')
	mqttc.on_message = on_message
	mqttc.on_connect = on_connect
	mqttc.on_subscribe = on_subscribe
	mqttc.on_publish = on_publish
	mqttc.on_log=on_log
	mqttc.connect(host, port, 60)
	mqttc.publish(topic, payload, 0)
	mqttc.subscribe(topic, 0)
	mqttc.loop_start()
	while True:
		while len(msgs) > 0:
			print("message: ",msgs.pop(0))
		while not q.empty():
			message = q.get()
			print("queue: ", message)
                        if message == 'quit':
                            break

	mqttc.loop_stop()

if __name__ == '__main__': client()

　　输出结果如下：

新线程中获得消息并存储，主线程中进行打印。实现新线程和主线程之间的通信。

你还可以在主线程中做一些其他可以做的事情。

https://blog.csdn.net/youshenmebutuo/article/details/79779387