SpringBoot2集成MQTT 实现消息的发布订阅

Posted 2021-09-04 风随心飞飞

最近项目中用到了，MQTT 实现 物联网行业的消息发布订阅，因为MQTT小巧，耗费流量少，在物联网中很受欢迎，在2G网络中就可以运行。MQTT 使用的是apache-apollo-1.7.1
**

安装：

1.下载apache-apollo-1.7.1
地址:http://archive.apache.org/dist/activemq/activemq-apollo/1.7.1/
2.解压，进入到D:\\java\\apache-apollo-1.7.1\\bin 目录下，执行命令

apollo.cmd create mybroker

3.进入到刚建好的目录，mybroker/bin目录，执行

apollo-broker.cmd run

成功！
看下目录结构：


主要配置文件：
etc/apollo.xml

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<!--
  Licensed to the Apache Software Foundation (ASF) under one or more
  contributor license agreements. See the NOTICE file distributed with
  this work for additional information regarding copyright ownership.
  The ASF licenses this file to You under the Apache License, Version
  2.0 (the "License"); you may not use this file except in compliance
  with the License. You may obtain a copy of the License at
  http://www.apache.org/licenses/LICENSE-2.0 Unless required by
  applicable law or agreed to in writing, software distributed under
  the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES
  OR CONDITIONS OF ANY KIND, either express or implied. See the
  License for the specific language governing permissions and
  limitations under the License.
-->

<!--
  For more information on how configure this file please
  reference:

  http://activemq.apache.org/apollo/versions/1.7.1/website/documentation/user-manual.html
  -->
<broker xmlns="http://activemq.apache.org/schema/activemq/apollo">

  <notes>
    The default configuration with tls/ssl enabled.
  </notes>

  <log_category console="console" security="security" connection="connection" audit="audit"/>

  
  <authentication domain="apollo"/>
  <!-- Give admins full access -->
  <access_rule allow="admins" action="*"/>
  <access_rule allow="*" action="connect" kind="connector"/>
  

  <virtual_host id="mybroker">
    <!--
      You should add all the host names that this virtual host is known as
      to properly support the STOMP 1.1 virtual host feature.
      -->
    <host_name>mybroker</host_name>
    <host_name>localhost</host_name>
    <host_name>127.0.0.1</host_name>

    <!-- Uncomment to disable security for the virtual host -->
    <!-- <authentication enabled="false"/> -->

    <!-- Uncomment to disable security for the virtual host -->
    <!-- <authentication enabled="false"/> -->
    <access_rule allow="users" action="connect create destroy send receive consume"/>
    

    <!-- You can delete this element if you want to disable persistence for this virtual host -->
    <leveldb_store directory="${apollo.base}/data"/>
    

  </virtual_host>

  <web_admin bind="http://127.0.0.1:61680"/>
  <web_admin bind="https://127.0.0.1:61681"/>

  <connector id="tcp" bind="tcp://0.0.0.0:61613" connection_limit="2000"/>
  <connector id="tls" bind="tls://0.0.0.0:61614" connection_limit="2000"/>
  <connector id="ws"  bind="ws://0.0.0.0:61623"  connection_limit="2000"/>
  <connector id="wss" bind="wss://0.0.0.0:61624" connection_limit="2000"/>

  <key_storage file="${apollo.base}/etc/keystore" password="password" key_password="password"/>

</broker>

etc/users.properties

## ---------------------------------------------------------------------------
## Licensed to the Apache Software Foundation (ASF) under one or more
## contributor license agreements.  See the NOTICE file distributed with
## this work for additional information regarding copyright ownership.
## The ASF licenses this file to You under the Apache License, Version 2.0
## (the "License"); you may not use this file except in compliance with
## the License.  You may obtain a copy of the License at
##
## http://www.apache.org/licenses/LICENSE-2.0
##
## Unless required by applicable law or agreed to in writing, software
## distributed under the License is distributed on an "AS IS" BASIS,
## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
## See the License for the specific language governing permissions and
## limitations under the License.
## ---------------------------------------------------------------------------

#
# The list of users that can login.  This file supports both plain text or
# encrypted passwords.  Here is an example what an encrypted password
# would look like:
#
# admin=ENC(Cf3Jf3tM+UrSOoaKU50od5CuBa8rxjoL)
#

admin=Admin@1234

其中用户名密码是 admin/Admin@1234
登录到后台管理网址：http://127.0.0.1:61680/console/index.html

输入刚才的admin/Admin@1234,进入到后台管理界面，可以看到里面的订阅主题等信息：

附上一段，mqtt 快速启动的 bat 代码

@echo off
title mqtt-server
set ENV_HOME="E:\\Program Files\\apache-apollo-1.7.1-windows-distro\\apache-apollo-1.7.1\\bin\\mybroker\\bin"
E:
color 0a
cd %ENV_HOME%
apollo-broker.cmd run
exit

与SpringBoot2 整合

废话少说，直接上代码：
1.Maven 中pom.mxl 配置

<!--MQtt-->
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-stream</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
        </dependency>

2.MqttConfig 配置类
此类里面提供了一些，mqtt 发布，订阅相关的配置

package com.dechnic.framework.config;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

/**
 * @description:
 * @author:houqd
 * @time: 2021/8/9 11:17
 */
@Configuration
@IntegrationComponentScan
public class MqttConfig {
    @Value("${spring.mqtt.username}")
    private String username;

    @Value("${spring.mqtt.password}")
    private String password;

    @Value("${spring.mqtt.url}")
    private String hostUrl;

    @Value("${spring.mqtt.client.id}")
    private String clientId;
    @Value("${spring.mqtt.client.subScribTopics}")
    private String[] subScribTopics;

    @Value("${spring.mqtt.default.topic}")
    private String defaultTopic;
    @Value("${spring.mqtt.completionTimeOut}")
    private int completionTimeOut ;   //连接超时

    @Bean
    public MqttConnectOptions getMqttConnectionOptions() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
        mqttConnectOptions.setServerURIs(new String[]{hostUrl});
        mqttConnectOptions.setKeepAliveInterval(2);
        return mqttConnectOptions;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectionOptions());
        return factory;
    }

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

    @Bean
     //ServiceActivator注解表明当前方法用于处理MQTT消息，inputChannel参数指定了用于接收消息信息的channel。
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound(){
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId,mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(defaultTopic);
        return messageHandler;
    }


    /**
     * 接收通道
     * @return
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
    //配置client,监听的topic
    @Bean
    public MessageProducer inbound(){
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId+"_inbound",mqttClientFactory(),subScribTopics);
        adapter.setCompletionTimeout(completionTimeOut);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }
    //通过通道获取数据
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
                String type = topic.substring(topic.lastIndexOf("/")+1, topic.length());
                if("hello".equalsIgnoreCase(topic)){
                    System.out.println("hello,XX,"+message.getPayload().toString());
                }else if("hello1".equalsIgnoreCase(topic)){
                    System.out.println("hello1,XX,"+message.getPayload().toString());
                }
            }
        };
    }


}

3.application.yml 配置

# 数据源配置
spring:
    datasource:
        type: com.alibaba.druid.pool.DruidDataSource
        driverClassName: com.mysql.cj.jdbc.Driver
        druid:
            # 主库数据源
            master:
                url: jdbc:mysql://localhost:3306/ds_db_ctl?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
                username: root
                password: 123456
            # 从库数据源
            slave:
                # 从数据源开关/默认关闭
                enabled: false
                url:
                username:
                password:
            # 初始连接数
            initialSize: 5
            # 最小连接池数量
            minIdle: 10
            # 最大连接池数量
            maxActive: 20
            # 配置获取连接等待超时的时间
            maxWait: 60000
            # 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
            timeBetweenEvictionRunsMillis: 60000
            # 配置一个连接在池中最小生存的时间，单位是毫秒
            minEvictableIdleTimeMillis: 300000
            # 配置一个连接在池中最大生存的时间，单位是毫秒
            maxEvictableIdleTimeMillis: 900000
            # 配置检测连接是否有效
            validationQuery: SELECT 1 FROM DUAL
            testWhileIdle: true
            testOnBorrow: false
            testOnReturn: false
            webStatFilter:
                enabled: true
            statViewServlet:
                enabled: true
                # 设置白名单，不填则允许所有访问
                allow:
                url-pattern: /druid/*
                # 控制台管理用户名和密码
                login-username: ruoyi
                login-password: 123456
            filter:
                stat:
                    enabled: true
                    # 慢SQL记录
                    log-slow-sql: true
                    slow-sql-millis: 1000
                    merge-sql: true
                wall:
                    config:
                        multi-statement-allow: true

    # redis 配置
    redis:
        # 地址
        host: localhost
        # 端口，默认为6379
        port: 6379
        # 数据库索引
        database: 0
        # 密码
        password: asdffdsa
        # 连接超时时间
        timeout: 10s
        lettuce:
            pool:
                # 连接池中的最小空闲连接
                min-idle: 0
                # 连接池中的最大空闲连接
                max-idle: 8
                # 连接池的最大数据库连接数
                max-active: 8
                # #连接池最大阻塞等待时间（使用负值表示没有限制）
                max-wait: -1ms

    mqtt:
        username: admin
        password: Admin@1234
        url: tcp://127.0.0.1:61613
        client:
            id: mqttId
            subScribTopics: Hello,Hello1
        default:
            topic: topic
        completionTimeOut: 3000

# 日志配置
logging:
    level:
        com.dechnic: info
        org.springframework: warn

4.MQTT 发布接口配置

package com.dechnic.framework.service;

import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateWay {
    void sendToMqtt(String data,@Header(MqttHeaders.TOPIC) String topic);
}

5.测试

package com.dechnic.apix.controller;

import com.dechnic.framework.service.MqttGateWay;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @description:
 * @author:houqd
 * @time: 2021/8/9 10:47
 */
@RestController
@RequestMapping("/test")
public class TestMqttController {
    @Autowired
    private MqttGateWay mqttGateWay;

    @RequestMapping("/sendMqtt.do")
    public String sendMqtt(String sendData,String topic){
        mqttGateWay.sendToMqtt(sendData,topic);
        return 以上是关于SpringBoot2集成MQTT 实现消息的发布订阅的主要内容，如果未能解决你的问题，请参考以下文章
 
 Spring Boot MQTT Too many publishes in progress错误的解决方案
 MQTT协议探究之消息集成
 Android使用MQTT订阅及发布消息（初步了解Mqtt以及实现Android操作mqtt服务）
 Spring Boot 集成 WebSocket，轻松实现信息推送！
 EMQX v4.4.5 发布：新增排他订阅及 MQTT 5.0 发布属性支持
 SpringBoot 集成 WebSocket，实现后台向前端推送信息