python实现MQTT协议（发布者，订阅者，topic）_使用python实现mqtt发布与订阅

python实现MQTT协议

一、简介

1.1 概述

本文章针对物联网MQTT协议完成python实现

1.2 环境

• Apache-apollo创建broker
• Python实现发布者和订阅者

1.3 内容

• MQTT协议架构说明 ：

• 利用仿真服务体会 MQTT协议

• 针对MQTT协议进行测试

任务1：MQTT协议应用场景

说明： MQ 遥测传输 (MQTT) 是轻量级基于代理的发布/订阅的消息传输协议，设计思想是开放、简单、轻量、 易于实现。这些特点使它适用于受限环境。该协议的特点有：

1 使用发布/订阅消息模式，提供一对多的消息发布，解除应用程序耦合。
2 对负载内容屏蔽的消息传输。
3 使用 TCP/IP 提供网络连接。

物联网应用场景:

协议角色分工：

客户端分为2种角色：发布者(Publisher)和订阅者(Subscriber)。
每一个发布者(Publisher)可以发送不同类型的消息，我们把消息的类型叫做主题（topic），
MQTT通信中的消息都属于某一个主题 ，而只有订阅了这个主题的订阅者(Subscriber)才能收到属于这个主题的消息。
发布者和订阅者不需要在意和知道对方的存在（不需要知道对方的IP和端口），也不需要直接与对方建立连接。因为通信中存在着一个叫代理 （MQTT broker）的第三种角色，也可以叫MQTT服务器（MQTT server）。 
发布者、订阅者只需要知道MQTT服务器的IP和端口即可，并和它直接建立连接通信。MQTT代理作为消息的 中转，它过滤所有接受到的消息，并按照一定的机制（MQTT标准规定是基于主题的消息过滤派发方式，而具 体的MQTT服务器软件也提供了其他的派发方式）分发它们，使得所有注册到MQTT代理的订阅者只接收到他 们订阅了的消息，而不会收到他不关心的消息。
当发布者发布一条消息的时候，他必须同时指定消息的主题和消息的负载。MQTT代理在收到发布者发过来的 消息时，无需访问消息负载，他只是访问消息的主题信息，然后根据这主题派发给订阅者。需要注意的是，一个客户端可以同时既当发布者又当订阅者。比如一个开发板连接了一盏LED灯，它可以发布灯的暗/亮状态 信息，也可以从其他节点订阅对灯的控制消息。
1
2
3
4
5
6

3.生产者（发布消息）和消费者（消耗消息-订阅者）模式理解

生产者:wifi设备采集各种物联网传感器比如温度重力传感器
消费者:客户端比如手机
原理如下:

任务2：搭建Mqtt协议服务 （broker）

前提:安装JDK和JAVA_HOME环境变量

1 下载Apollo服务器 地址 http://archive.apache.org/dist/activemq/activemq-apollo/1.7.1/

2 进入bin目录命令行 输入：

D:\softwares\apache-apollo-1.7.1\bin\apollo.cmd create jwbroker
1

3:broker\etc\apollo.xml文件下是配置服务器信息的文件
初始默认帐号是admin，密码password;

4:启动命令行: (以一个实例为单位进行创建的)

进入... jwbroker创建实例的\bin\ 目录,
在CMD输入命令apollo-broker.cmd run，可以使用TAB键自动补全，运行后输出信息如下:验证:
MQTT服务器TCP连接端口: tcp://0.0.0.0:61613
后台web管理页面:https://127.0.0.1:61681/或http://127.0.0.1:61680/
1
2
3
4

出现如下图表示启动成功

安装mqtt需要的包：

pip install paho-mqtt
1

发布者publish创建：

import time
from paho.mqtt import publish
#源码中只需要知道   ip + 端口 + 订阅的主题
HOST ="127.0.0.1"
PORT =61613

def on_connect(client,userdata, flags,rc):
    print("Connected with result code" + str(rc))
    client.subscribe("jw-temperature") # 发布主题

def on_message(client,userdata,msg):
    print(msg.topc +  "消息发送！" + msg.payload.decode("utf-8"))

if __name__ == '__main__':
    print("消息发布！----我是一个发布者：正在给设备和传感器发布主题-----")
    client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
    for i in range(20):
        time.sleep(2)
        publish.single("lightChange","现在天黑了", qos = 1, hostname = HOST, port = PORT, client_id = client_id,
                       auth = {'username': "admin", 'password': "password"})
        print("已发送"+str(i+1)+"条消息")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

订阅者light_subcribe创建：

import paho.mqtt.client as mqtt
import time

#源码中只需要知道   ip + 端口 + 订阅的主题
HOST ="127.0.0.1"
PORT =61613

'''
The callback for when the client receives a CONNACK response from the server
客户端接收到服务器端的确认连接请求时，回调on_connect服务端发送CONNACK报文响应从客户端收到的CONNECT报文。
服务端发送给客户端的第一个报文必须是CONNACK [MQTT-3.2.0-1].
'''
def on_connect(client,userdata, flags,rc):
    print("Connected with result code" + str(rc))
    '''
    Subscribing in on_connect() means that if we lose the connection 
    and  reconnect then subscriptions wil be renewed(恢复、续订).'''
    client.subscribe("lightChange") # 订阅主题

'''
The callback for when a PUBLISH message is received from the server.
客户端接收到服务器向其传输的消息时，
回调on_messagePUBLISH控制报文是指从客户端向服务端或者服务端向客户端传输一个应用消息。
'''
def on_message(client,userdata,msg):
    print(msg.topic+ msg.payload.decode("utf-8") +  "，回调消息：收到收到！我已经接收到发布者的消息，并且打开了光传感器" )

def client_loop():
    '''
    注意，client_id是必须的，并且是唯一的。否则可能会出现如下错误
    [WinError 10054] 远程主机强迫关闭了一个现有的连接
    '''

    client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
    client = mqtt.Client(client_id) # Client_id 不能重复，所以使用当前时间
    client.username_pw_set("admin","password") # 必须设置，否则会返回 /Connected with result code 4/
    client.on_connect = on_connect
    client.on_message = on_message

    '''
    拥塞回调:处理网络流量，调度回调和重连接。
    Blocking call that processes network traffic, 
    dispatches callbacks and handles reconnecting.
    Other loop*() functions are available that give a threaded interface and amanual- interface...I
    '''
    try:
        client.connect(HOST,PORT,60)
        client.loop_forever()
    except KeyboardInterrupt:
        client.disconnect()

if __name__ == '__main__':
    print("手电筒打开----我是一个订阅者：需要消费主题-----")
    client_loop()
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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

订阅者phone_subcribe创建：

import paho.mqtt.client as mqtt
import time

#源码中只需要知道   ip + 端口 + 订阅的主题
HOST ="127.0.0.1"
PORT =61613
'''
The callback for when the client receives a CONNACK response from the server
客户端接收到服务器端的确认连接请求时，回调on_connect服务端发送CONNACK报文响应从客户端收到的CONNECT报文。
服务端发送给客户端的第一个报文必须是CONNACK [MQTT-3.2.0-1].

'''
def on_connect(client,userdata, flags,rc):
    print("Connected with result code" + str(rc))
    '''
    Subscribing in on_connect() means that if we lose the connection 
    and  reconnect then subscriptions wil be renewed(恢复、续订).
    '''
    client.subscribe("lightChange") # 订阅主题


'''
The callback for when a PUBLISH message is received from the server.
客户端接收到服务器向其传输的消息时，
回调on_messagePUBLISH控制报文是指从客户端向服务端或者服务端向客户端传输一个应用消息。
'''
def on_message(client,userdata,msg):
    print(msg.topic  + msg.payload.decode("utf-8")+  "，回调消息：收到收到！我已经接收到发布者的消息并给用户反馈手电筒已经打开")

def client_loop():
    '''
    注意，client_id是必须的，并且是唯一的。否则可能会出现如下错误
    [WinError 10054] 远程主机强迫关闭了一个现有的连接
    '''

    client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
    client = mqtt.Client(client_id) # Client_id 不能重复，所以使用当前时间
    client.username_pw_set("admin","password") # 必须设置，否则会返回 /Connected with result code 4/
    client.on_connect = on_connect
    client.on_message = on_message

    '''
    拥塞回调:处理网络流量，调度回调和重连接。
    Blocking call that processes network traffic, 
    dispatches callbacks and handles reconnecting.
    Other loop*() functions are available that give a threaded interface and amanual- interface...I
    '''
    try:
        client.connect(HOST,PORT,60)
        client.loop_forever()
    except KeyboardInterrupt:
        client.disconnect()

if __name__ == '__main__':
    print("手机启动----我是一个订阅者：需要消费主题-----")
    client_loop()
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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

演示：分别运行publish和light_subcribe和phone_subcribe

publish：

light_subcribe：

phone_subcribe：