SpringBoot集成MQTT协议

发布时间:2024年01月03日

简介

MQTT 可以被解释为一种低开销,低带宽占用的即时通讯协议,可以用较少的代码和带宽为远程设备连接提供实时可靠的消息服务,它适用于硬件性能低下的远程设备以及网络状况糟糕的环境下,因此 MQTT 协议在 IoT(Internet of things,物联网),小型设备应用,移动应用等方面有较广泛的应用。

优点:代码量少,开销低,带宽占用小,即时通讯协议。

MQTT原理

实现mqtt协议需要客户端和服务器端通讯完成,在通讯中,mqtt协议中有三种身份:发布者(publish),代理(broker)(服务器),订阅者(subscribe)。其中,消息的发布者和订阅者都是客户端。消息代理是服务器,消息发布者可以同时是订阅者,传输过程如下如所示。

image.png

有别于传统的客户端/服务器通讯协议,MQTT协议并不是端到端的,消息传递通过代理,包括会话(session)也不是建立在发布者和订阅者之间,而是建立在端和代理之间。代理解除了发布者和订阅者之间的耦合。

除了发布者和订阅者之间传递普通消息,代理还可以为发布者处理保留消息和遗愿消息,并可以更改服务质量(QoS)等级。

MQTT主题

1、主题层级分隔符“/”

用于分割主题的每个层级,为主题名提供一个分层结构。如主题:

china/anhui
china/anhui/hefei

2、多层通配符“#”

用于匹配主题中任意层级的通配符。如主题:china/#

china/anhui
china/anhui/hefei
china/anhui/hefei/shushan

3、单层通配符“+”

加号是只能用于单个主题层级匹配的通配符。如主题:

china/+      只能匹配 china/anhui
china/+/+/shushan     能匹配china/anhui/hefei/shushan

4、通配符“$”

通配符“$”表示匹配一个字符,只要不是放在主题的最开头,即:

$xx
/$xx
/xx$

实战应用

对接协议部分内容

image.png

SpringBoot集成Mqtt协议

本篇文章就不重点介绍MQTT相关的内容了,主要学会怎么运用到实际开发工作中。

  1. 添加依赖
<dependency>
   <groupId>org.springframework.integration</groupId>
   <artifactId>spring-integration-mqtt</artifactId>
</dependency>
  1. 添加配置类
@Configuration
@ConfigurationProperties(prefix="spring.mqtt")
public class MqttProperties {
    private String url;
    private String username;
    private String password;
    private String clientId;

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getClientId() {
        return clientId;
    }

    public void setClientId(String clientId) {
        this.clientId = clientId;
    }
}
@Configuration
@ConditionalOnBean(MqttProperties.class)
public class MqttConfiguration {

    @Autowired
    private MqttProperties mqttProperties;

    @Autowired
    private MqttMessageHandler mqttMessageHandler;

    /**
     * 创建MqttPahoClientFactory,设置MQTT Broker连接属性,如果使用SSL验证,也在这里设置。
     *
     * @return factory
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();

        // 设置代理端的URL地址,可以是多个
        options.setServerURIs(new String[]{mqttProperties.getUrl()});
        options.setUserName(mqttProperties.getUsername());
        options.setPassword(mqttProperties.getPassword().toCharArray());
        options.setKeepAliveInterval(120);
        factory.setConnectionOptions(options);
        return factory;
    }

    /**
     * 入站通道
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    /**
     * 入站
     */
    @Bean
    public MessageProducer inbound() {
        // Paho客户端消息驱动通道适配器,主要用来订阅主题
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttProperties.getClientId() + "-consumer",
                mqttClientFactory(),
                MqttConstants.UP_DEVICE_STATE,
                MqttConstants.UP_DEVICE_STATUS,
                MqttConstants.UP_DEVICE_EVENT,
                MqttConstants.UP_DEVICE_SET_ACK,
                MqttConstants.UP_DEVICE_GET_ACK,
                MqttConstants.UP_DEVICE_CAPTURE_REPORT,
                MqttConstants.UP_DEVICE_CONTROL_QUERY_ACK,
                MqttConstants.UP_DEVICE_CONTROL_ACK);
        // Paho消息转换器
        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
        adapter.setConverter(defaultPahoMessageConverter);
        adapter.setCompletionTimeout(5000);
        // 设置QoS
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    /**
     * ServiceActivator注解表明:当前方法用于处理MQTT消息,inputChannel参数指定了用于消费消息的channel。
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return mqttMessageHandler;
    }

    /**
     * 出站通道
     */
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * 出站
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler outbound() {
        // 发送消息和消费消息Channel可以使用相同MqttPahoClientFactory
        MqttPahoMessageHandler mqttPahoMessageHandler = new MqttPahoMessageHandler(
                mqttProperties.getClientId() + "-producer", mqttClientFactory());
        // 如果设置成true,即异步,发送消息时将不会阻塞。
        mqttPahoMessageHandler.setAsync(true);
        // 设置默认QoS
        mqttPahoMessageHandler.setDefaultQos(1);
        // Paho消息转换器
        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
        mqttPahoMessageHandler.setConverter(defaultPahoMessageConverter);
        return mqttPahoMessageHandler;
    }
}
  1. 消息处理类
@Service
public class MqttMessageHandler implements MessageHandler {

    private final Logger logger = LoggerFactory.getLogger(MqttMessageHandler.class);

    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        try {
            String payload = message.getPayload().toString();
            String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
            logger.info("接受来自mqtt的订阅信息,topic:{}", topic);
            //离线上报
            if (topic.matches(".+/offline")) {
                statusReport(payload);
            }
            //状态上报
            else if (topic.matches(".+/status")) {
                deviceInfoReport(payload);
            } 
            //事件上报
            else if (topic.matches(".+/eventReport")) {
                eventReport(payload);
            } else {
                logger.info("主题topic:{},负载payload:{}", topic, payload);
            }
        } catch (Exception e) {
            logger.error("handleMessage 接受mqtt订阅消息异常:", e);
        }
    }
    
      /**
     * 设备状态上报
     *
     * @param payload
     */
    private void statusReport(String payload) {
        OfflineReport offlineReport = JSONUtil.toBean(payload, OfflineReport.class);
        logger.info("收到设备状态信息上报:{}", offlineReport);
        //...省略
    }

 //...省略
   
}
/**
 * 消息发送
 */
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {

    /**
     * 定义重载方法,用于消息发送
     *
     * @param payload
     */
    void sendToMqtt(String payload);

    /**
     * 指定topic进行消息发送
     *
     * @param topic
     * @param payload
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);

    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);

    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, byte[] payload);
}

源码地址:https://gitee.com/jiangwang001/springboot/tree/master/cy-cabinet-adapter

小结

本文简单介绍了一下MQTT协议的基本知识,在实际工作中,MQTT通常应用于物联网、智能家居等设备和应用程序之间的通信。在嵌入式领域,MQTT已经占据着无法替代的分量,因为大多数的嵌入式设备,都需要这样的协议进行数据交互。

2024一起加油~

文章来源:https://blog.csdn.net/jiang_wang01/article/details/135359414
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。