消息中间件(MQTT)

• 一、简介
• 二、概念
• 三、SpringBoot+Maven实现发布、订阅
• 四、MQTT代理
• 五、开发遇到的问题

    MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上。MQTT适用于移动端场景，侧重多语言多平台的海量设备接入，如遥感数据、汽车、智能家居等。

1. QoS（服务质量）
     用于保证消息稳定传输的机制，包括消息应答、存储和重传。QoS0：至多一次，存在消息漏收的情况；QoS1：至少一次，有应答机制；QoS2：确保只有一次，开销大。
     QoS降级：实际的订阅QoS = Min(Publisher发布消息指定的QoS，Subscriber订阅与Broker协商的QoS)。即：实际的订阅QoS是消息过程中小的那个QoS值。
   
   

1. pom中添加依赖

    org.springframework.integration
    spring-integration-core
    5.2.8.RELEASE


    org.springframework.boot
    spring-boot-starter-integration
    2.3.0.RELEASE


    org.springframework.integration
    spring-integration-stream
    5.2.8.RELEASE


    org.springframework.integration
    spring-integration-mqtt
    5.2.8.RELEASE

2. yml中添加配置

spring:
  mqtt:
    username: username
    password: password
    url: tcp://ip:port
    clientId: mqttclient
    topic: topic1
    completionTimeout: 2000

3. Java实现
    为使结构清晰，分为以下五部分。其中，前三个为配置类。
    MqttbaseConfig：配置Mqtt公用部分，主要是提供Mqtt客户端Bean。
    MqttInConfig：配置入站消息通道及相关处理Bean。
    MqttOutConfig：配置出站消息通道及相关处理Bean。
    MqttMessageReceiver：入站消息处理。
    MqttMessageSender：出站消息发送接口。
   
   

 市面上有相当多的高质量MQTT代理，其中mosquitto是一个开源的轻量级的C实现，完全兼容了MQTT 3.1和MQTT 3.1.1。

订阅命令：
mosquitto_sub -d -t 'topic1'

发布命令：
mosquitto_pub -d -t 'topic1' -m 'test123'

1. 提示已断开连接; retrying…
    断开的原因是clientId相同，可以拼上当前时间戳确保clientId不同。发布端可以固定clientId，消费端需确保clientId不同，否则会出现接受不到消息的情况。
2. 发布消息后，订阅服务器漏收
    可能与Qos属性未设置有关。添加handler.setDefaultQos(1);可解决。