1 import paho.mqtt.client as mqtt
 2 
 3 import logging
 4 
 5 class MqttClient(mqtt.Client):
 6 
 7     def initClient(self, mqttServer, mqttPort, username, password, timeout=10000):
 8         logging.basicConfig(level=logging.DEBUG)
 9 
10         self.mqttServer = mqttServer
11         self.mqttPort = mqttPort
12         self.username_pw_set(username, password=password)
13 
14         self.connect(self.mqttServer, self.mqttPort, timeout)  # keeplive仅为10000秒
15         self.on_connect = self.on_connect
16 
17     def getClient(self):
18         return self.client
19 
20     def on_connect(self, client, userdata, flags, rc):
21         linkAddr = client.mqttServer + ":"+ str(client.mqttPort)
22         if rc == 0:
23             logging.info("与mqtt服务器："+ linkAddr +"连接成功")
24         elif rc == 1:
25             logging.error("协议版本错误")
26         elif rc == 2:
27             logging.error("无效的客户端标识")
28         elif rc == 3:
29             logging.error("服务端无法使用")
30         elif rc == 4:
31             logging.error("与mqtt服务器连接失败: 错误的用户名或密码 ")
32         elif rc == 5:
33             logging.error("登录用户未经授权 ")
34         else:
35             logging.error("与mqtt服务器：%s 连接返回异常结果：%s " % (linkAddr, str(rc)))
36 
37     def on_subscribe(self, client, userdata, mid, granted_qos):
38         logging.info("订阅成功: " + str(mid) + " " + str(granted_qos))
39 
40     def on_publish(self, client, userdata, mid):
41         logging.info("OnPublish, mid: " + str(mid))

 1 #!/usr/bin/python
 2 import datetime
 3 import json
 4 import logging
 5 import time
 6 
 7 from mqttService.mqttClient import MqttClient
 8 
 9 #------------------客户端1--------------------
10 # ======================================================
11 
12 
13 # 服务器地址
14 mqttServer = "192.168.1.103"
15 # 通信端口
16 mqttPort = 1883
17 # 订阅主题
18 devTopic = '/devices/dev1'
19 responseDevTopic= '/7.0/dev1'
20 
21 
22 # 接收客户端2响应信息
23 def on_message(client, userdata, message):
24     curtime = datetime.datetime.now()
25     strcurtime = curtime.strftime("%Y-%m-%d %H:%M:%S")
26     logging.info("%s: 接收 %s 响应信息：主题：%s 内容：%s" %(strcurtime, mqttServer+":"+str(mqttPort), message.topic, str(message.payload, encoding = "utf-8")))
27 
28 
29 if __name__ == '__main__':
30     logging.basicConfig(level=logging.DEBUG)
31 
32     username = "UserName1"
33     password = "PassWord1"
34 
35     client = MqttClient()
36     client.initClient(mqttServer, mqttPort, username, password)
37     client.subscribe(responseDevTopic, qos=0)  #订阅主题
38     client.on_message = on_message
39 
40     msg = 'hello world'
41     for i in range(1000):
42         client.publish(devTopic, payload=msg , qos=0)  #发布信息
43         time.sleep(4)

　　　　client.loop_forever()  # 持续连接

 #!/usr/bin/python
 import datetime
 import json
 import logging
 
 from mqttService.mqttClient import MqttClient
 
  #------------------客户端2-------------------
  
 # 订阅主题
 devTopic = '/devices/#'
 # 发布主题

 responseDevTopicPrefix = '/7.0/'
 
 def on_message(client, userdata, message):
     curtime = datetime.datetime.now()
     strcurtime = curtime.strftime("%Y-%m-%d %H:%M:%S")
     recvMsg = message.payload
 
     # 获取发送设备的设备ID
     topicArr = str(message.topic).split("/")
     topicArr = [item for item in filter(lambda x: x != '', topicArr)] # 去除空串
     deviceId = topicArr[1]

     logging.info("%s: 接收硬件装置 %s 信息：主题: %s 内容: %s" %(strcurtime, deviceId, message.topic, recvMsg))
 
     # 发送响应回硬件终端
     client.publish(responseDevTopicPrefix +deviceId, payload="收到数据"+recvMsg, qos=0)if __name__ == '__main__':
     logging.basicConfig(level=logging.DEBUG)
 
     mqttAddr = '192.168.1.103'
     mqttPort = 1883

     username = "UserName2"
     password = "PassWord2"
     # 与客户端1通信
     client = MqttClient()
     client.initClient(mqttAddr, mqttPort, username, password)
     client.subscribe(devTopic, qos=0)
     client.on_message = on_message
     client.loop_start() # 开始监听
    # 阻塞主程序
     while True:
        pass