SpringBoot整合mqtt协议，实现订阅topic并将消息存储到mysql数据库

所需依赖如下：

<dependencies>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-web</artifactId>

        </dependency>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-devtools</artifactId>

            <scope>runtime</scope>

            <optional>true</optional>

        </dependency>

        <dependency>

            <groupId>mysql</groupId>

            <artifactId>mysql-connector-java</artifactId>

            <scope>runtime</scope>

        </dependency>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-configuration-processor</artifactId>

            <optional>true</optional>

        </dependency>

        <dependency>

            <groupId>org.projectlombok</groupId>

            <artifactId>lombok</artifactId>

            <optional>true</optional>

        </dependency>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-test</artifactId>

            <scope>test</scope>

            <exclusions>

                <exclusion>

                    <groupId>org.junit.vintage</groupId>

                    <artifactId>junit-vintage-engine</artifactId>

                </exclusion>

            </exclusions>

        </dependency>

        <dependency>

            <groupId>org.springframework.integration</groupId>

            <artifactId>spring-integration-mqtt</artifactId>

        </dependency>

        <dependency>

            <groupId>com.alibaba</groupId>

            <artifactId>fastjson</artifactId>

            <version>1.2.72</version>

        </dependency>

        <dependency>

            <groupId>com.baomidou</groupId>

            <artifactId>mybatis-plus-boot-starter</artifactId>

            <version>3.4.2</version>

        </dependency>

        <dependency>

            <groupId>com.github.pagehelper</groupId>

            <artifactId>pagehelper-spring-boot-starter</artifactId>

            <version>1.3.1</version>

        </dependency>

    </dependencies>

示例代码如下：

package com.cyl.mqttdemo.config;

import com.alibaba.fastjson.JSONObject;

import com.cyl.mqttdemo.entity.SysData;

import com.cyl.mqttdemo.service.SysDataService;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;

import org.eclipse.paho.client.mqttv3.MqttMessage;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

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.support.DefaultPahoMessageConverter;

import org.springframework.messaging.MessageChannel;

import org.springframework.messaging.MessageHandler;

/**

 * @author cyl

 * @time 2022/8/4

 */

@Configuration

public class MqttReceiveConfig {

    @Autowired

    private SysDataService dataService;

    @Bean

    public MqttConnectOptions getMqttConnectOptions() {

        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();

        //mqttConnectOptions.setUserName("6KvPBqHX");

        mqttConnectOptions.setUserName("FZgZdPNfgjwgncaT");

        //mqttConnectOptions.setPassword("riEtkxce".toCharArray());

        mqttConnectOptions.setPassword("SuiCSeQ3omdybpGt".toCharArray());

        //String[] hostArray = new String[]{"tcp://mqtt.lnxall.com:3883"};

        String[] hostArray = new String[]{"tcp://173.16.200.25:3883"};

        mqttConnectOptions.setServerURIs(hostArray);

        mqttConnectOptions.setKeepAliveInterval(2);

        mqttConnectOptions.setCleanSession(false);

        return mqttConnectOptions;

    }

    @Bean

    public MqttPahoClientFactory mqttClientFactory() {

        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();

        factory.setConnectionOptions(getMqttConnectOptions());

        return factory;

    }

    // 点对点路由通道

    @Bean

    public MessageChannel mqttInputChannel() {

        return new DirectChannel();

    }

    // 定义入站消息适配器

    @Bean

    public MessageProducer inbound() {

        // 定义监听的topic名称ClientId等信息

        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("java-mqtt", mqttClientFactory(), "/sys/00000000/#"); // 94fbRocF

        adapter.setCompletionTimeout(5000);

        adapter.setConverter(new DefaultPahoMessageConverter());

        adapter.setQos(1);

        adapter.setOutputChannel(mqttInputChannel());

        return adapter;

    }

    //  可调用Spring的Bean来处理消息，并将处理后的结果输出到指定的消息通道

    @Bean

    @ServiceActivator(inputChannel = "mqttInputChannel")

    public MessageHandler handler() {

        return message -> {

            MqttMessage mqttMessage = JSONObject.parseObject(message.getPayload().toString(), MqttMessage.class);

            // 这里可以根据不同的topic name处理不同的业务逻辑

            String topic = message.getHeaders().get("mqtt_receivedTopic").toString();

            String[] split = topic.split("/");

            if (split.length==5&& "service".equals(split[4])) {

                System.out.println("------------------------------------");

                System.out.println("topic:" + topic);

                String payload = message.getPayload().toString();

                System.out.println("payload:" + payload);

                System.out.println("------------------------------------");

                SysData data = JSONObject.parseObject(payload, SysData.class);

                dataService.save(data);

            }

        };

    }

}

SpringBoot整合mqtt协议，实现订阅topic并将消息存储到mysql数据库的相关教程结束。