mosquitto（MQTT）服务器搭建和基本使用

一、安装

搭建一个mqtt服务器，这里我们采用mosquitto

1、 下载地址：https://mosquitto.org/download/

 

2、 选择windows：https://mosquitto.org/files/binary/win64/mosquitto-2.0.18-install-windows-x64.exe

 

3、 右键安装即可

默认的安装路径在这里：C:\Program Files\mosquitto

 

 

二、修改配置文件

1、 进入mosquitto的安装目录，找到配置文件

 C:\Program Files\mosquitto\ mosquitto.conf

 

2、 修改mosquitto指定的监听端口

 listener 8899

 

 

3、 关闭mosquitto的匿名登录功能

 

allow_anonymous false

 

 

 

4、 修改配置用户的配置文件路径（这里要用绝对路径）

 

password_file D:\other\mosquitto\pwd.emample

 

 

三、启动

1、 进入windows的service。启动mqtt broker服务

 

2、命令行启动：mosquitto.exe -c mosquitto.conf -v

 

四、使用

1、用python写一个订阅者的代码

import paho.mqtt.client as mqtt
import ssl

context = ssl.create_default_context()
# 加载证书
context.load_default_certs()

# 将证书转换为常量
cert_data = ssl.DER_cert_to_PEM_cert(context.get_ca_certs()[0])


broker = '127.0.0.1'
port = 7788
topic = "AIOT"

# 当客户端和服务端连接成功后执行的回调函数
def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")

    # 当连接成功后，客户端订阅一个主题，并消费该主题的消息
    client.subscribe(topic=topic)

# 当客户端订阅的主题收到消息的时候执行的回调函数
def on_message(client, userdata, msg):
    # client 那个客户端
    # userdata 用户数据
    # msg 实际的消息对象
    print(msg.topic+" "+str(msg.payload))

# 实例化一个客户端，实现mqtt协议版本3.1、3.1.1、5.0的客户端类，是一个主要的与MQTT服务端通信的类
client = mqtt.Client()

# 实例化的时候可以传递参数进去
# client_id="",
# 唯一的客户端ip字符串，当前连接mqtt服务器时候使用，如果cliendid长度为0，则会自动随机生产一个clientid，这种情况下，clean_session参数必须是true


# clean_session=None,
# 决定clint类型的参数，如果为true，当client失去连接时候中间人将删除所有关于这个client的信息，如果为false，则说明这是一个永久的client，当失去连接的时候，发布的消息和队列里的消息
# 都会被保持。
# 注意，当失去连接的时候clinet永远不会抛弃自己需要发布的消息，当调用connetc（）/reconnect（）会重新发送这些消息
# userdata=None,

# 用户定义的任何类型的数据将被传输为userdata，后面可以使用user_data_set()来更新这个参数
# protocol=MQTTv311,
# client使用的mqtt版本
# transport="tcp",
# 设置通信机制，websocket或者tcp、一个是websocket传输，一个tcp传输
# reconnect_on_failure=True
# 连接失败是否重新连接

# 客户端绑定回调方法,其实这里定义的很多回调函数
client.on_connect = on_connect
client.on_message = on_message
# client.on_disconnect
#
# client.on_publish




# 客户端连接mqtt broker服务器
client.connect(bind_address=broker, bind_port=port, keepalive=60)
# client.connect_async()

# 客户端和mqtt broker服务器断开连接

client.disconnect()
# 永久执行，保持和mqtt broker服务端的长期通信
client.loop_forever()

　　

我们当然也可以用命令行的方式启动订阅者，可以通过help来查询使用方法

 

mosquitto_sub.exe –help

 

2、发布者代码

import paho.mqtt.client as mqtt
import time

broker = '127.0.0.1'
port = 8899
topic = "AIOT"

def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")

client = mqtt.Client()
client.on_connect = on_connect
client.connect(broker, port, 60)
for i in range(60):
    client.publish(topic, payload=i, qos=0, retain=False)
    print(f"send {i} to a/b{topic}")
    time.sleep(1)

client.loop_forever()

　　

当然我们可以使用命令行的方式启用生产者，可以通过help来查询使用方法

 

mosquitto_pub.exe –help

 

3、通过证书连接mqtt broker

import paho
import ssl

mqttc = paho.mqtt.client.Client('my_client')
...
# 构建一个SSL上下文
SSL_CTX = {
    'ssl_port': 8884,
    'ca': "/ ca.crt",
    'client_cert': '/client/client.crt',
     'client_key':" /client/client.key",
     'cert_reqs': None,
    'tls_version': None,
    'ciphers': None,
    'insecure': False  # 关闭insecure选项
}
# 设置TLS参数
mqttc.tls_set(SSL_CTX['ca'],
              certfile=SSL_CTX['client_cert'],
              keyfile=SSL_CTX['client_key'],
              cert_reqs=SSL_CTX['cert_reqs'],
              tls_version=SSL_CTX['tls_version'],
              ciphers=SSL_CTX['ciphers'])
# ca_certs=None, certfile=None, keyfile=None, cert_reqs=None, tls_version=None, ciphers=None, keyfile_password=Non
# 要求验证服务端证书中域名与mqtt连接创建时输入的broker域名一致
mqttc.tls_insecure_set(SSL_CTX['insecure'])