0
0

SpringBoot整合MQTT（MqttClient）

一、SpringBoot整合MQTT

创建项目，引入 MQTT依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><version>2.3.12.RELEASE</version></dependency><!-- spring-integration-mqtt依赖 --><dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId><version>6.1.2</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.28</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.75</version></dependency>

1、yml配置文件

在 application.yml文件中，定义 MQTT连接信息。

## MQTT 基本连接参数 ##mqtt:host: tcp://192.168.xxx.xxx:1883#  host: tcp://broker.emqx.io:1883userName: admin
  passWord: xxxxxx
  qos:1clientId: ClientId_local #ClientId_local必须唯一。timeout:10# 超时时间keepalive:30# 保持连接时间clearSession:true# 清除会话(设置为false,断开连接，重连后使用原来的会话 保留订阅的主题，能接收离线期间的消息)topic1: A/b/#  # 通配符主题只能用于订阅，不能用于发布。+：表示单层通配符，#：表示多层通配符topic2: A/abc
  topic3: ABC

2、MQTT配置类

创建一个 MqttConfig配置类，并获取配置文件的 MQTT的连接参数。创建 MyMqttClient类注入Spring。

@Slf4j@ConfigurationpublicclassMqttConfig{@Value("${mqtt.host}")publicString host;@Value("${mqtt.username}")publicString username;@Value("${mqtt.password}")publicString password;@Value("${mqtt.clientId}")publicString clientId;@Value("${mqtt.timeout}")publicint timeOut;@Value("${mqtt.keepalive}")publicint keepAlive;@Value("${mqtt.clearSession}")publicboolean clearSession;@Value("${mqtt.topic1}")publicString topic1;@Value("${mqtt.topic2}")publicString topic2;@Value("${mqtt.topic3}")publicString topic3;@Bean//注入SpringpublicMyMqttClientmyMqttClient(){MyMqttClient myMqttClient =newMyMqttClient(host, username, password, clientId, timeOut, keepAlive, clearSession);for(int i =0; i <10; i++){try{
                myMqttClient.connect();// 这里可以订阅主题，推荐放到 MqttCallbackExtended.connectComplete方法中//myMqttClient.subscribe("ABC", 1);return myMqttClient;}catch(MqttException e){
                log.error("== MqttConfig ==> MQTT connect exception, connect time = {}", i);try{Thread.sleep(2000);}catch(InterruptedException e1){
                    e1.printStackTrace();}}}return myMqttClient;}}

3、MQTT 客户端封装类

创建 MQTT 客户端封装类MyMqttClient。对 MQTT Broker进行操作。

@Slf4jpublicclassMyMqttClient{/**
     * MQTT Broker 基本连接参数，用户名、密码为非必选参数
     */privateString host;privateString username;privateString password;privateString clientId;privateint timeout;privateint keepalive;privateboolean clearSession;/**
     * MQTT 客户端
     */privatestaticMqttClient client;publicstaticMqttClientgetClient(){return client;}publicstaticvoidsetClient(MqttClient client){MyMqttClient.client = client;}publicMyMqttClient(String host,String username,String password,String clientId,int timeOut,int keepAlive,boolean clearSession){this.host = host;this.username = username;this.password = password;this.clientId = clientId;this.timeout = timeOut;this.keepalive = keepAlive;this.clearSession = clearSession;}/**
     * 设置 MQTT Broker 基本连接参数
     *
     * @param username
     * @param password
     * @param timeout
     * @param keepalive
     * @return
     */publicMqttConnectOptionssetMqttConnectOptions(String username,String password,int timeout,int keepalive){MqttConnectOptions options =newMqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        options.setCleanSession(clearSession);
        options.setAutomaticReconnect(true);return options;}/**
     * 连接 MQTT Broker，得到 MqttClient连接对象
     */publicvoidconnect()throwsMqttException{if(client ==null){
            client =newMqttClient(host, clientId,newMemoryPersistence());// 设置回调
            client.setCallback(newMyMqttCallback(MyMqttClient.this));}// 连接参数MqttConnectOptions mqttConnectOptions =setMqttConnectOptions(username, password, timeout, keepalive);if(!client.isConnected()){
            client.connect(mqttConnectOptions);}else{
            client.disconnect();
            client.connect(mqttConnectOptions);}
        log.info("== MyMqttClient ==> MQTT connect success");//未发生异常，则连接成功}/**
     * 发布，默认qos为0，非持久化
     *
     * @param pushMessage
     * @param topic
     */publicvoidpublish(String pushMessage,String topic){publish(pushMessage, topic,0,false);}/**
     * 发布消息
     *
     * @param pushMessage
     * @param topic
     * @param qos
     * @param retained:留存
     */publicvoidpublish(String pushMessage,String topic,int qos,boolean retained){MqttMessage message =newMqttMessage();
        message.setPayload(pushMessage.getBytes());
        message.setQos(qos);
        message.setRetained(retained);MqttTopic mqttTopic =MyMqttClient.getClient().getTopic(topic);if(null== mqttTopic){
            log.error("== MyMqttClient ==> topic is not exist");}MqttDeliveryToken token;//Delivery:配送synchronized(this){//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁，分析见文章最后补充try{
                token = mqttTopic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
                token.waitForCompletion(1000L);}catch(MqttPersistenceException e){
                e.printStackTrace();}catch(MqttException e){
                e.printStackTrace();}}}/**
     * 订阅某个主题，qos默认为0
     *
     * @param topic
     */publicvoidsubscribe(String topic){subscribe(topic,0);}/**
     * 订阅某个主题
     *
     * @param topic
     * @param qos
     */publicvoidsubscribe(String topic,int qos){try{MyMqttClient.getClient().subscribe(topic, qos);}catch(MqttException e){
            e.printStackTrace();}
        log.info("== MyMqttClient ==> 订阅主题成功：topic = {}， qos = {}", topic, qos);}/**
     * 取消订阅主题
     *
     * @param topic 主题名称
     */publicvoidcleanTopic(String topic){if(client !=null&& client.isConnected()){try{
                client.unsubscribe(topic);}catch(MqttException e){
                e.printStackTrace();}}else{
            log.error("== MyMqttClient ==> 取消订阅失败！");}
        log.info("== MyMqttClient ==> 取消订阅主题成功：topic = {}", topic);}}

**

说明：

**

  • MqttClient: 同步调用客户端，使用阻塞方法通信。
  • MqttClientPersistence: 代表一个持久的数据存储，用于在传输过程中存储出站和入站的信息，使其能够传递到指定的 QoS。
  • MqttConnectOptions: 连接选项，用于指定连接的参数，下面列举一些常见的方法。 - setUserName: 设置用户名- setPassword: 设置密码- setCleanSession: 设置是否清除会话- setKeepAliveInterval: 设置心跳间隔- setConnectionTimeout: 设置连接超时时间- setAutomaticReconnect: 设置是否自动重连

4、MqttClient回调类

创建一个 MqttClient回调类MyMqttCallback。

@Slf4jpublicclassMyMqttCallbackimplementsMqttCallbackExtended{//手动注入privateMqttConfig mqttConfig =SpringUtils.getBean(MqttConfig.class);privateMyMqttClient myMqttClient;publicMyMqttCallback(MyMqttClient myMqttClient){this.myMqttClient = myMqttClient;}/**
     * MQTT Broker连接成功时被调用的方法。在该方法中可以执行 订阅系统约定的主题（推荐使用）。
     * 如果 MQTT Broker断开连接之后又重新连接成功时，主题也需要再次订阅，将重新订阅主题放在连接成功后的回调方法中比较合理。
     *
     * @param reconnect
     * @param serverURI MQTT Broker的url
     */@OverridepublicvoidconnectComplete(boolean reconnect,String serverURI){String connectMode = reconnect ?"重连":"直连";
        log.info("== MyMqttCallback ==> MQTT 连接成功，连接方式：{}，serverURI：{}", connectMode, serverURI);//订阅主题
        myMqttClient.subscribe(mqttConfig.topic1,1);
        myMqttClient.subscribe(mqttConfig.topic2,1);
        myMqttClient.subscribe(mqttConfig.topic3,1);List<String> topicList =newArrayList<>();
        topicList.add(mqttConfig.topic1);
        topicList.add(mqttConfig.topic2);
        topicList.add(mqttConfig.topic3);
        log.info("== MyMqttCallback ==> 连接方式：{}，订阅主题成功，topic：{}", connectMode, topicList);}/**
     * 丢失连接，可在这里做重连
     * 只会调用一次
     *
     * @param throwable
     */@OverridepublicvoidconnectionLost(Throwable throwable){
        log.error("== MyMqttCallback ==> connectionLost 连接断开，5S之后尝试重连: {}", throwable.getMessage());long reconnectTimes =1;while(true){try{if(MyMqttClient.getClient().isConnected()){//判断已经重新连接成功  需要重新订阅主题 可以在这个if里面订阅主题  或者 connectComplete（方法里面）  看你们自己选择
                    log.warn("== MyMqttCallback ==> mqtt reconnect success end  重新连接  重新订阅成功");return;}
                reconnectTimes +=1;
                log.warn("== MyMqttCallback ==> mqtt reconnect times = {} try again...  mqtt重新连接时间 {}", reconnectTimes, reconnectTimes);MyMqttClient.getClient().reconnect();}catch(MqttException e){
                log.error("== MyMqttCallback ==> mqtt断连异常", e);}try{Thread.sleep(5000);}catch(InterruptedException e1){}}}/**
     * 接收到消息（subscribe订阅的主题消息）时被调用的方法
     *
     * @param topic
     * @param mqttMessage
     * @throws Exception 后得到的消息会执行到这里面
     */@OverridepublicvoidmessageArrived(String topic,MqttMessage mqttMessage)throwsException{
        log.info("== MyMqttCallback ==> messageArrived 接收消息主题: {}，接收消息内容: {}", topic,newString(mqttMessage.getPayload()));/**
         * 根据订阅的主题分别处理业务。可以通过if-else或者策略模式来分别处理不同的主题消息。
         *///topic1主题if(topic.equals("ABC")){Map maps =(Map)JSON.parse(newString(mqttMessage.getPayload(),StandardCharsets.UTF_8));//TODO 业务处理//doSomething1(maps);
            log.info("== MyMqttCallback ==> messageArrived 接收消息主题: {}，{}业务处理消息内容完成", topic,"TodoService1");}//topic2主题if(topic.equals("A/b/1qaz")){Map maps =(Map)JSON.parse(newString(mqttMessage.getPayload(),StandardCharsets.UTF_8));//TODO 业务处理//doSomething2(maps);
            log.info("== MyMqttCallback ==> messageArrived 接收消息主题: {}，{}业务处理消息内容完成", topic,"TodoService2");}}/**
     * 消息发送（publish）完成时被调用的方法
     *
     * @param iMqttDeliveryToken
     */@OverridepublicvoiddeliveryComplete(IMqttDeliveryToken iMqttDeliveryToken){
        log.info("== MyMqttCallback ==> deliveryComplete 消息发送完成，Complete= {}", iMqttDeliveryToken.isComplete());}}

MqttCallback类方法说明：

  • connectionLost(Throwable cause): 连接丢失时被调用
  • messageArrived(String topic, MqttMessage message): 接收到消息时被调用
  • deliveryComplete(IMqttDeliveryToken token): 消息发送完成时被调用

**

MqttCallbackExtended类方法说明：该类继承MqttCallback类

**

  • connectComplete(boolean reconnect, String serverURI): 连接丢失时被调用

4.1 SpringUtils工具类

@ComponentpublicclassSpringUtilsimplementsBeanFactoryPostProcessor,ApplicationContextAware{/**
     * Spring应用上下文环境
     */privatestaticConfigurableListableBeanFactory beanFactory;privatestaticApplicationContext applicationContext;@OverridepublicvoidpostProcessBeanFactory(ConfigurableListableBeanFactory beanFactory)throwsBeansException{SpringUtils.beanFactory = beanFactory;}@OverridepublicvoidsetApplicationContext(ApplicationContext applicationContext)throwsBeansException{SpringUtils.applicationContext = applicationContext;}/**
     * 获取对象
     *
     * @param name
     * @return Object 一个以所给名字注册的bean的实例
     * @throws org.springframework.beans.BeansException
     */@SuppressWarnings("unchecked")publicstatic<T>TgetBean(String name)throwsBeansException{return(T) beanFactory.getBean(name);}/**
     * 获取类型为requiredType的对象
     *
     * @param clz
     * @return
     * @throws org.springframework.beans.BeansException
     */publicstatic<T>TgetBean(Class<T> clz)throwsBeansException{T result =(T) beanFactory.getBean(clz);return result;}/**
     * 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true
     *
     * @param name
     * @return boolean
     */publicstaticbooleancontainsBean(String name){return beanFactory.containsBean(name);}/**
     * 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）
     *
     * @param name
     * @return boolean
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     */publicstaticbooleanisSingleton(String name)throwsNoSuchBeanDefinitionException{return beanFactory.isSingleton(name);}/**
     * @param name
     * @return Class 注册对象的类型
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     */publicstaticClass<?>getType(String name)throwsNoSuchBeanDefinitionException{return beanFactory.getType(name);}/**
     * 如果给定的bean名字在bean定义中有别名，则返回这些别名
     *
     * @param name
     * @return
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     */publicstaticString[]getAliases(String name)throwsNoSuchBeanDefinitionException{return beanFactory.getAliases(name);}/**
     * 获取aop代理对象
     *
     * @param invoker
     * @return
     */@SuppressWarnings("unchecked")publicstatic<T>TgetAopProxy(T invoker){return(T)AopContext.currentProxy();}/**
     * 获取当前的环境配置，无配置返回null
     *
     * @return 当前的环境配置
     */publicstaticString[]getActiveProfiles(){return applicationContext.getEnvironment().getActiveProfiles();}/**
     * 获取当前的环境配置，当有多个环境配置时，只获取第一个
     *
     * @return 当前的环境配置
     */publicstaticStringgetActiveProfile(){finalString[] activeProfiles =getActiveProfiles();if(activeProfiles ==null){returnnull;}return activeProfiles[0];}}

到此，Springboot 通过 MqttClient整合操作 MQTT Broker就可以了。

二、操作MQTT

我们在 service层创建一个 MqttService类，业务通过 MqttService类来统一操作 MqttClient。

1、自定义发送消息载体类

这里创建一个 MyXxxMqttMsg类，来约定发送消息的载体类格式。

@DatapublicclassMyXxxMqttMsgimplementsSerializable{privatestaticfinallong serialVersionUID =-8303548938481407659L;/**
     * MD5值：MD5_lower(content + timestamp)
     */privateString md5;/**
     * 消息内容
     */privateString content ="";/**
     * 时间戳
     */privateLong timestamp;}

2、MqttService类

1）接口：

publicinterfaceMqttService{/**
     * 添加订阅主题
     *
     * @param topic 主题名称
     */voidaddTopic(String topic);/**
     * 取消订阅主题
     *
     * @param topic 主题名称
     */voidremoveTopic(String topic);/**
     * 发布主题消息内容
     *
     * @param msgContent
     * @param topic
     */voidpublish(String msgContent,String topic);}

2）实现类：

@ServicepublicclassMqttServiceImplimplementsMqttService{@AutowiredprivateMyMqttClient myMqttClient;@OverridepublicvoidaddTopic(String topic){
        myMqttClient.subscribe(topic);}@OverridepublicvoidremoveTopic(String topic){
        myMqttClient.cleanTopic(topic);}@Overridepublicvoidpublish(String msgContent,String topic){//MyXxxMqttMsg 转JsonMyXxxMqttMsg myXxxMqttMsg =newMyXxxMqttMsg();
        myXxxMqttMsg.setContent(msgContent);
        myXxxMqttMsg.setTimestamp(System.currentTimeMillis());// TODO Md5值
        myXxxMqttMsg.setMd5(UUID.randomUUID().toString());String msgJson =JSON.toJSONString(myXxxMqttMsg);//发布消息
        myMqttClient.publish(msgJson, topic);}

3、controller类

创建一个 MyMqttController类，来操作一下 MQTT。

@RestController@RequestMapping("/mqtt")@Api(value ="MyMqttController", tags ={"MQTT相关操作接口"})publicclassMyMqttController{@AutowiredprivateMqttService mqttService;@GetMapping("/addTopic")@ApiOperation(value ="添加订阅主题接口")publicvoidaddTopic(String topic){
        mqttService.addTopic(topic);}@GetMapping("/removeTopic")@ApiOperation(value ="取消订阅主题接口")publicvoidremoveTopic(String topic){
        mqttService.removeTopic(topic);}@PostMapping("/removeTopic")@ApiOperation(value ="发布主题消息内容接口")publicvoidremoveTopic(String msgContent,String topic){
        mqttService.publish(msgContent, topic);}}

在这里插入图片描述

订阅和取消主题操作：MQTTX发布了一个主题消息。

在这里插入图片描述

发布通配符主题消息：

在这里插入图片描述

– 求知若饥，虚心若愚。

标签： SpringBoot 整合MQTT MqttClient
本文转载自: https://blog.csdn.net/qq_42402854/article/details/132791347
版权归原作者 Charge8 所有， 如有侵权，请联系我们删除。
登录后发布评论

“SpringBoot整合MQTT（MqttClient）”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

机器人技术在安全领域的重要作用

安全学习记录——网络篇（一）

图腾柱PFC：HP1010为您的电动两轮车之旅提供绿色，高效，安全的动力

单元测试数据库回滚问题

这才是单元测试，之前我们都错了

【C语言基础】：字符函数和字符串函数

ChatGPT-Next-Web SSRF漏洞+XSS漏洞复现(CVE-2023-49785)

文章导航