Java通过RabbitMQ实现MQTT通信

1. 前言：

简单介绍下mqtt

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IOT）。

1. 发布方（Publisher）将消息发送到 Broker（MQTT服务器）；
2. Broker 接收到消息以后，检查下都有哪些订阅方订阅了此类消息，然后将消息发送到这些订阅方；
3. 订阅方（Subscriber）从 Broker 获取该消息。

MQTT它只是一种协议，支持MQTT协议的消息中间件产品非常多，下边的也只是其中的一部分

• Mosquitto
• Eclipse Paho
• RabbitMQ
• Apache ActiveMQ
• HiveMQ
• JoramMQ
• ThingMQ
• VerneMQ
• Apache Apollo
• emqttd Xively
• IBM Websphere .....

服务端需要通过mqtt推送消息到安卓客户端，这里使用RabbitMQ做为broker，这里也可以选择其他的mq作为mqtt的broker

选的测试工具为mqttbox，

1.1. 参考文档

1.1.1. rabbitmq安装

1.1.2. mqtt发送和订阅消息资料参考

1.2. 开启MQTT协议

参考

RabbitMQ安装好以后，需要开启 mqtt 协议

默认情况下RabbitMQ是不开启MQTT 协议的，所以需要我们手动的开启相关的插件，而RabbitMQ的MQTT 协议分为两种。

第一种 rabbitmq_mqtt 提供与后端服务交互使用，对应端口1883。

rabbitmq-plugins enable rabbitmq_mqtt

第二种 rabbitmq_web_mqtt 提供与前端交互使用，对应端口15675。

rabbitmq-plugins enable rabbitmq_web_mqtt

2. 代码实现

2.1. 引入Maven依赖

按照spring官方的文档，引入spring-integration-mqtt理论上就够了

2.1.1. 引入依赖

2.1.1.1. 引入mqtt的相关依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.7.RELEASE</version>
        <relativePath/>
    </parent>
    <groupId>com.test</groupId>
    <artifactId>mqtt-push</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>mqtt-push</name>

    <description>demo project for Spring Boot mqtt push</description>

    <properties>
        <java.version>1.8</java.version>
        <paho.client.mqttv3.version>1.2.5</paho.client.mqttv3.version>
    </properties>
  
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <!--mqtt依赖包-->
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
        </dependency>

<!--        <dependency>-->
<!--            <groupId>org.eclipse.paho</groupId>-->
<!--            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>-->
<!--            <version>${paho.client.mqttv3.version}</version>-->
<!--        </dependency>-->

        <!-- lombok工具包 -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>

    </dependencies>

</project>

2.2. mqtt公共的配置信息

包括clientId和serverClientId，默认的topic信息, 以及连接rabbitmq的用户名和密码

2.2.1. clientId的唯一性

注意，我们在开发测试阶段clientId直接在代码中写死了，而且服务都是单实例部署，并没有暴露出什么问题。然而在生产环境内侧的时候，由于服务是多实例集群部署，如果每台服务器的clientId都是一样的，出现下边的奇怪问题。同一时间内只能有一个客户端能拿到消息，其他客户端不但不能消费消息，而且还在不断的掉线重连：Lost connection: 已断开连接; retrying...。

同样的，这个在 上也提到了

clientId，由业务系统分配，需要保证每个 tcp 连接都不一样，保证全局唯一，如果不同的客户端对象（tcp 连接）使用了相同的 clientId 会导致连接异常断开

这里可以使用随机数或者分布式id来生成clientId

public class MQ4IoTProducerDemo {
    /**
     * MQ4IOT clientId，由业务系统分配，需要保证每个 tcp 连接都不一样，保证全局唯一，如果不同的客户端对象（tcp 连接）使用了相同的 clientId 会导致连接异常断开。
     * clientId 由两部分组成，格式为 GroupID@@@DeviceId，其中 groupId 在 MQ4IOT 控制台申请，DeviceId 由业务方自己设置，clientId 总长度不得超过64个字符。
     */
    private String clientId = "GID_XXXXX@@@XXXXX";

}

2.2.2. yaml配置信息

server:
  port: 8888

# spring:
#   main:
#     allow-bean-definition-overriding: true


mqtt-push:
    # clientId的前缀
    clientId: mqtt_client_
    # 通过通配符 "+", 订阅主题, 就可以接收所有传感器发送的温度数据了
    # 多个的话通过,号分割
    defaultTopic: sensor/+/temperature
    # clientId的前缀
    serverClientId: mqtt_server_
    # 这边替换为自己的ip(或者域名)
    # 多个的话通过,号分割
    servers: tcp://127.0.0.1:1883
    # 访问rabbitmq的用户名
    username: guest
    # 访问rabbitmq的密码
    password: guest

2.2.3. Mqtt通用的配置信息

package com.mqttpush.config;

import lombok.Data;
import lombok.Getter;
import lombok.Setter;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.stereotype.Component;

/**
 * mqtt的连接配置
 * @author
 */
@Data
@Component
@IntegrationComponentScan
@ConfigurationProperties(prefix = "mqtt-push")
public class MqttConfig {

    /**
     * 服务地址
     */
    private String servers;

    /**
     * 客户端id
     */
    private String clientId;

    /**
     * 服务端id
     */
    private String serverClientId;

    /**
     * 默认主题
     */
    private String defaultTopic;

    /**
     * 用户名
     */
    private String username;

    /**
     * 密码
     */
    private String password;

    /**
     * 创建MqttPahoClientFactory，设置MQTT Broker连接属性，如果使用SSL验证，也在这里设置
     *
     * @return MqttPahoClientFactory
     */
    @Bean(name = "mqttClientFactory")
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConnectOptions());
        return factory;
    }

    @Bean(name = "mqttConnectOptions")
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        //断开后，是否自动连接
        options.setAutomaticReconnect(true);
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
        // 把配置里的 cleanSession 设为false，客户端掉线后 服务器端不会清除session，
        // 当重连后可以接收之前订阅主题的消息。当客户端上线后会接受到它离线的这段时间的消息
        options.setCleanSession(false);
        // 设置连接的用户名
        options.setUserName(username);
        // 设置连接的密码
        options.setPassword(password.toCharArray());
        options.setServerURIs(servers.split(","));
        // 设置超时时间 单位为秒
        options.setConnectionTimeout(10);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
        options.setKeepAliveInterval(20);
        // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
        //        options.setWill("willTopic", WILL_DATA, 2, false);
        return options;
    }
}

springboot项目中集成框架，有消息入站通道（用来接收消息）和出站通道（用来发送消息）

2.3. 发布消息配置

package com.mqttpush.config;

import com.mqttpush.constant.MqttConstant;
import com.mqttpush.util.RandomUtil;
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.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;

/**
 * mqtt 消息发布配置
 * @author
 */
@Configuration
public class MqttProducerConfig {

    @Resource
    private MqttConfig mqttConfig;

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * ServiceActivator注解表明：当前方法用于处理MQTT消息，inputChannel参数指定了用于消费消息的channel。
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = MqttConstant.CHANNEL_NAME_OUT)
    public MessageHandler mqttOutbound() {
        // clientId相同导致客户端间相互竞争消费
        // 同一时间内只能有一个客户端能拿到消息, 其他客户端不但不能消费消息，而且还在不断的掉线重连：Lost connection: 已断开连接; retrying...。
        MqttPahoMessageHandler messageHandler
                = new MqttPahoMessageHandler(mqttConfig.getServerClientId() + MqttConstant.MESSAGE_HANDLER_CLIENT_ID_PRODUCER + RandomUtil.getRandomStr(),
                                                    mqttConfig.mqttClientFactory());
        messageHandler.setAsync(true);
        // MQTT 提供了三种服务质量（QoS），在不同网络环境下保证消息的可靠性。
        // QoS 0：消息最多传送一次。如果当前客户端不可用，它将丢失这条消息。
        // QoS 1：消息至少传送一次。
        // QoS 2：消息只传送一次。
        messageHandler.setDefaultQos(1);
        messageHandler.setDefaultTopic(mqttConfig.getDefaultTopic());
        return messageHandler;
    }
}

2.4. 订阅消息配置

通信是双向的, 服务端既可以发布消息，也可以订阅消息，如果不需要的话，此代码不需要引入

package com.mqttpush.config;

import com.mqttpush.constant.MqttConstant;
import com.mqttpush.util.RandomUtil;
import lombok.extern.slf4j.Slf4j;
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.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import javax.annotation.Resource;

/**
 * mqtt 消息订阅配置
 *
 * @author
 */
@Slf4j
@Configuration
public class MqttSubscriberConfig {

    @Resource
    private MqttConfig mqttConfig;

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        // clientId相同导致客户端间相互竞争消费
        // 同一时间内只能有一个客户端能拿到消息, 其他客户端不但不能消费消息，而且还在不断的掉线重连：Lost connection: 已断开连接; retrying...。
        MqttPahoMessageDrivenChannelAdapter adapter
                    = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId() + MqttConstant.MESSAGE_HANDLER_CLIENT_ID_CONSUMER + RandomUtil.getRandomStr(),
                                                                    mqttConfig.mqttClientFactory(),
                                                                    mqttConfig.getDefaultTopic());
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(2);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    /**
     * 消息订阅, 处理收到的消息
     */
    @Bean
    @ServiceActivator(inputChannel = MqttConstant.CHANNEL_NAME_IN)
    public MessageHandler mqttInMessageHandler() {
        return message -> {
            try {
                // 消息体
                String payload = message.getPayload().toString();
                log.info("接收到消息, 内容: {}", payload);

                // byte[] bytes = (byte[]) message.getPayload(); // 收到的消息是字节格式
                // 消息的topic
                String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString();
                log.info("接收到消息, topic: {}", topic);

                // 根据主题分别进行消息处理。
                if (topic.matches(".+/sensor")) { // 匹配：1/sensor
                    String sensorSn = topic.split("/")[0];
                    log.info("传感器" + sensorSn + ": 的消息： " + payload);
                } else if (topic.equals("collector")) {
                    log.info("采集器的消息：" + payload);
                } else {
                    log.info("丢弃消息：主题[" + topic  + "]，负载：" + payload);
                }
            } catch (MessagingException ex) {
                //logger.info(ex.getMessage());
            }
        };

    }
}

2.5. 常量类

package com.mqttpush.constant;

/**
 * mqtt常量类
 *
 * @author
 * @since 2024-04-27
 */
public class MqttConstant {

    /**
     * mqtt发布者信道名称
     */
    public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";


    /**
     * mqtt接收信道名称
     */
    public static final String CHANNEL_NAME_IN = "mqttInputChannel";

    /**
     * mqtt消息发布者, serverClientId的前缀
     */
    public static final String MESSAGE_HANDLER_CLIENT_ID_PRODUCER = "producer";

    /**
     * mqtt消息接收者, clientId的前缀
     */
    public static final String MESSAGE_HANDLER_CLIENT_ID_CONSUMER = "consumer";
}

2.6. 发布消息

使用@MessagingGateway注解发送消息

package com.mqttpush.producer;

import com.mqttpush.constant.MqttConstant;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;

/**
 * rabbitmq mqtt协议网关接口
 */
@MessagingGateway(defaultRequestChannel = MqttConstant.CHANNEL_NAME_OUT)
public interface MqttGateway {

    void sendMessage2Mqtt(String data);

    void sendMessage2Mqtt(String data, @Header(MqttHeaders.TOPIC) String topic);

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

2.7. 控制器和测试类

package com.mqttpush.controller;


import com.mqttpush.dto.ReqSendMsgDTO;
import com.mqttpush.producer.MqttGateway;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;

/**
 *
 * mqtt发送消息控制器
 *
 * @author
 */
@RestController
@RequestMapping("/mqtt-push")
public class MqttController {

    @Resource
    private MqttGateway mqttGateway;

    @PostMapping("/sendMessage")
    public String sendMqtt(@RequestBody ReqSendMsgDTO reqSendMsgDTO) {
        mqttGateway.sendMessage2Mqtt(reqSendMsgDTO.getTopic(), reqSendMsgDTO.getPayload());
        return "SUCCESS";
    }
}

测试类

package com.mqttpush;

import com.mqttpush.producer.MqttGateway;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import javax.annotation.Resource;

@SpringBootTest
class MqttPushApplicationTests {

    @Resource
    private MqttGateway mqttGateway;

    @Test
    void contextLoads() {
    }

    @Test
    public void sendMqtt() {
        mqttGateway.sendMessage2Mqtt("sensor/s123/temperature", "我是中间通配符topic的内容");
    }

}

3. 测试

3.1. MQTTBOX连接配置

3.2. MQTTBOX 发送和订阅配置

左边为发布消息，右边为订阅消息

3.3. 发送消息

可以使用MQTTBOX或者PostMan发送消息

用MQTTBOX或者服务端接收消息

3.3.1. 使用Postman发送消息

3.3.1.1. 服务器订阅消息

3.3.1.2. MQTTBOX订阅消息

3.3.2. MQTTBOX发送消息

3.3.2.1. MQTTBOX订阅消息

3.3.2.2. 代码订阅消息