如何在Spring Boot中使用MQTT

2024-04-02 19:04:59 410人浏览安东尼

Python 官方文档：入门教程 => 点击学习

摘要

目录为什么选择MQTT MQtT, 启动! 使用方式 Client模式创建工厂类创建工具类 spring Integration 总结为什么选择MQTT MQTT的定义相信很

为什么选择MQTT

MQTT的定义相信很多人都能讲的头头是道,本文章也不讨论什么高大上的东西,旨在用最简单直观的方式让每一位刚接触的同行们可以最快的应用起来

先从使用MQTT需要什么开始分析:

消息服务器
不同应用/设备之间的频繁交互
可能涉及一对多的消息传递

根据上面列举的这三点,我们大概可以了解到, MQTT最适合的场景是消息做为系统的重要组成部分,且参与着系统关键业务逻辑的情形

MQTT, 启动!

既然决定使用它,我们首先要研究的是如何让MQTT正常工作,毕竟它不是简单的在Maven里加入个依赖就完事的

我们总共需要干如下两件事:

下载EMQX消息服务器, 作为broker
在maven中引入依赖


<dependency>  
    <groupId>org.springframework.integration</groupId>  
    <artifactId>spring-integration-mqtt</artifactId>  
    <version>5.3.2.RELEASE</version>  
</dependency>

完成上面两步后, 启动EMQX服务器, 正式进入我们的MQTT旅途

使用方式

在Spring Boot中使用MQTT的代码, 笔者总结了如下两种方式:

使用spring-integration的消息通道概念
使用传统的Client客户端概念

第一种会产生一定程度的心智负担,但在笔者成功搭配(抄袭+造轮子)自动注册后, 比后者要方便许多

在介绍具体代码之前, 我们先简单整理下使用中最常见的概念:

主题: MQTT消息的主要传播途径, 我们向主题发布消息, 订阅主题, 从主题中读取消息并进行业务逻辑处理, 主题是消息的通道
生产者: MQTT消息的发送者, 他们向主题发送消息
消费者: MQTT消息的接收者, 他们订阅自己需要的主题, 并从中获取消息
broker: 消息转发器, 消息是通过它来承载的, EMQX就是我们的broker, 在使用中我们不用关心它的具体实现

其实, MQTT的使用流程就是: 生产者给主题发消息->broker进行消息的传递->订阅该主题的消费者拿到消息并进行相应的业务逻辑

Client模式

本模式和传统的数据库链接,Redis链接基本一致,有开发经验的小伙伴们可以很轻松的驾驭,我们需要考虑的就是如果创建对应的工厂,是单例模式,还是原型,亦或是造个池子呢?

我们使用单例模式来进行本次的介绍

创建工厂类

首先, 我们创造一个工厂(就不承认设计模式中毒)


public class MqttFactory {  

    private static MqttProperties configuration;  
    
    private static MqttClient client;  
 
    
    public static MqttClient getInstance() {    
        if (client == null) {      
            init();    
        }    
        return client;  
    }  
    
    
    public static void init() {    
        try {      
            client = new MqttClient(configuration.getAddress(), "client-" + System.currentTimeMillis());      
            // MQTT配置对象
            MqttConnectOptions options = new MqttConnectOptions();      
            // 设置自动重连, 其它具体参数可以查看MqttConnectOptions
            options.setAutomaticReconnect(true);      
            if (!client.isConnected()) {        
            client.connect(options);      
            }    
        } catch (MqttException e) {      
            LOGGER.error(String.fORMat("MQTT: 连接消息服务器[%s]失败", configuration.getAddress()));    
        }  
    }
    
}

关于MQTT的具体配置可以查看MqttConnectOptions, 在这里就不做说明了

多嘴一句, 文档永远比某些博客给力!!!

创建工具类

接下来, 我们创建MqttUtil, 用于消息的发送以及主题的订阅


public class MqttUtil {  

    
    public static void send(String topic, Object data) {    
        // 获取客户端实例
        MqttClient client = MqttFactory.getInstance();    
        ObjectMapper mapper = new ObjectMapper();    
        try {
            // 转换消息为JSON字符串
            String json = mapper.writeValueAsString(data);      
            client.publish(topic, new MqttMessage(json.getBytes(StandardCharsets.UTF_8)));    
        } catch (JsonProcessingException e) {      
            LOGGER.error(String.format("MQTT: 主题[%s]发送消息转换json失败", topic));    
        } catch (MqttException e) {      
            LOGGER.error(String.format("MQTT: 主题[%s]发送消息失败", topic));    
        }  
    }
    
    
    public static void subscribe(String topic, IMqttMessageListener listener) {  
        MqttClient client = MqttFactory.getInstance();  
        try {    
            client.subscribe(topic, listener);  
        } catch (MqttException e) {    
            LOGGER.error(String.format("MQTT: 订阅主题[%s]失败", topic));  
        }
    }
    
}

相信小伙伴们注意到了IMqttMessageListener这个东西, 我们只需要创建一个监听类, 实现IMqttMessageListener接口, 就可以处理消息啦, 代码如下:


public class MessageListener implements IMqttMessageListener {  

    
    @Override  
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {   
        LOGGER.info(String.format("MQTT: 订阅主题[%s]发来消息[%s]", topic, new String(mqttMessage.getPayload())));  
    }
    
    public static void main(String[] args) {  
        //订阅主题test01, 使用MessageListener来处理它的消息
        MqttUtil.subscribe("test01", new MessageListener());
    }

}

无论是发送还是订阅,是不是都很好理解?

舒服的事情结束后, 带来的是无尽的折磨和空虚, 来吧, 让我们挑战下心智负担大的第二种模式!