Docker MQTT安装使用教程_服务器知识

MQTT简介

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是IBM开发的一个即时通讯协议，有可能成为物联网的重要组成部分。该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当做传感器和制动器（比如通过Twitter让房屋联网）的通信协议。

Docker安装RabbitMQ配置MQTT

使用RabbitMQ作为MQTT服务端，Eclipse Paho作为客户端。宿主机系统为ubuntu16.04

Docker下载镜像

docker pull daocloud.io/library/rabbitmq:3.7.4

启动RabbitMQ

1	`docker run` `-d` `--hostname my-rabbit` `--name some-rabbit` `-p` `15672:15672` `-p` `5672:5672` `-p` `1883:1883` `-p` `15675:15675` `daocloud.io/library/rabbitmq:3.7.4`

注意映射容器端口

15672 是rabbitmq management管理界面默认访问端口
5672 是amqp默认端口
1883 是mqtt tcp协议默认端口
15675 是web_mqtt websocket协议默认端口

启用插件

默认安装后我们需要手动开启rabbitmq_management插件，rabbitmq_mqtt插件和rabbitmq_web_mqtt插件。

执行如下三条命令

									docker exec <容器ID> rabbitmq-plugins enable rabbitmq_management

									docker exec <容器ID> rabbitmq-plugins enable rabbitmq_mqtt

									docker exec <容器ID> rabbitmq-plugins enable rabbitmq_web_mqtt

当然你也可以写个脚本start.sh，复制到容器中

									/usr/sbin/rabbitmq-plugins enable rabbitmq_management

									/usr/sbin/rabbitmq-plugins enable rabbitmq_mqtt

									/usr/sbin/rabbitmq-plugins enable rabbitmq_web_mqtt

进入容器执行这个脚本。

sh start.sh

开放宿主机端口

									firewall-cmd --zone=public --add-port=15672/tcp --permanent

									firewall-cmd --zone=public --add-port=5672/tcp --permanent

									firewall-cmd --zone=public --add-port=1883/tcp --permanent

									firewall-cmd --zone=public --add-port=15675/tcp --permanent

									firewall-cmd --reload

Python MQTT客户端实现

安装python包

pip install paho-mqtt

发送数据demo（消费者）

									# 使用前需要启动hbase和thrift服务器

									# 启动hbase在cd /usr/local/hbase下bin/start-hbase.sh  默认端口为 60000

									# 启动thrift服务器cd /usr/local/hbase/bin执行./hbase-daemon.sh start thrift  默认端口为9090

									import sys

									import os

									dir_common = os.path.split(os.path.realpath(__file__))[0] + '/../'

									sys.path.append(dir_common)  # 将根目录添加到系统目录,才能正常引用common文件夹

									import argparse  #

									import logging

									import time,datetime

									from common.py_log import init_logger,init_console_logger

									from common.config import *

									from common.py_hbase import PyHbase

									import time,json

									from common.py_rabbit import Rabbit_Consumer

									import paho.mqtt.client as mqtt

									import time

									HOST = "192.168.2.46"

									PORT = 1883

									def client_loop():

									  client_id = time.strftime('%Y%m%d%H%M%S',time.localtime(time.time()))

									  client = mqtt.Client(client_id)  # ClientId不能重复，所以使用当前时间

									  client.username_pw_set("guest", "guest") # 必须设置，否则会返回「Connected with result code 4」

									  client.on_connect = on_connect

									  client.on_message = on_message

									  client.connect(HOST, PORT, 60)

									  client.loop_forever()

									def on_connect(client, userdata, flags, rc):

									  print("Connected with result code "+str(rc))

									  client.subscribe("test")

									def on_message(client, userdata, msg):

									  print(msg.topic+" "+msg.payload.decode("utf-8"))

									if __name__ == '__main__':

									  client_loop()

接收数据demo（生产者）

									import sys

									import os

									dir_common = os.path.split(os.path.realpath(__file__))[0] + '/../'

									sys.path.append(dir_common)  # 将根目录添加到系统目录,才能正常引用common文件夹

									import paho.mqtt.client as mqtt

									import time

									HOST = "192.168.2.46"

									PORT = 1883

									def client_loop():

									  client_id = time.strftime('%Y%m%d%H%M%S',time.localtime(time.time()))

									  client = mqtt.Client(client_id)  # ClientId不能重复，所以使用当前时间

									  client.username_pw_set("guest", "guest") # 必须设置，否则会返回「Connected with result code 4」

									  client.on_connect = on_connect

									  client.on_message = on_message

									  client.connect(HOST, PORT, 60)

									  client.loop_forever()

									def on_connect(client, userdata, flags, rc):

									  print("Connected with result code "+str(rc))

									  client.subscribe("test")

									def on_message(client, userdata, msg):

									  print(msg.topic+" "+msg.payload.decode("utf-8"))

									if __name__ == '__main__':

									  client_loop()

生产者demo

									# import paho.mqtt.client as mqtt

									import paho.mqtt.publish as publish

									import time

									HOST = "192.168.2.46"

									PORT = 1883

									def on_connect(client, userdata, flags, rc):

									  print("Connected with result code "+str(rc))

									  client.subscribe("test")

									def on_message(client, userdata, msg):

									  print(msg.topic+" "+msg.payload.decode("utf-8"))

									if __name__ == '__main__':

									  client_id = time.strftime('%Y%m%d%H%M%S',time.localtime(time.time()))

									  # client = mqtt.Client(client_id)  # ClientId不能重复，所以使用当前时间

									  # client.username_pw_set("guest", "guest") # 必须设置，否则会返回「Connected with result code 4」

									  # client.on_connect = on_connect

									  # client.on_message = on_message

									  # client.connect(HOST, PORT, 60)

									  # client.publish("test", "你好 MQTT", qos=0, retain=False) # 发布消息

									  publish.single("test", "你好 MQTT", qos = 1,hostname=HOST,port=PORT, client_id=client