程序员的成长之路JAVA · ARCHITECTURE · GROWTH

外观

Java通过RabbitMQ实现MQTT通信

约 2859 字大约 10 分钟

2025-01-16

1. 前言：

简单介绍下mqtt

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

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

Java通过RabbitMQ实现MQTT通信配图（1）

Java通过RabbitMQ实现MQTT通信配图（2）

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安装

Windows安装RabbitMQ详细教程

如何在Windows中的Rabbitmq如何启动

RabbitMQ 和 MQTT 实践 web消息实时推送（这个小红点）

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

spring官方mqtt示例文档

Java实现MQTT通信

我也没想到 springboot + rabbitmq 做智能家居，会这么简单

1.2. 开启MQTT协议

参考Windows安装RabbitMQ详细教程

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

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

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

rabbitmq-plugins enable rabbitmq_mqtt

1.2. 开启MQTT协议配图（1）

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

rabbitmq-plugins enable rabbitmq_web_mqtt

1.2. 开启MQTT协议配图（2）

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…。

2.2.1. clientId的唯一性配图

同样的，这个在 阿里云的mqtt开发手册上也提到了

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.1. MQTTBOX连接配置配图

3.2. MQTTBOX 发送和订阅配置

左边为发布消息，右边为订阅消息3.2. MQTTBOX 发送和订阅配置配图

3.3. 发送消息

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

用MQTTBOX或者服务端接收消息

3.3.1. 使用Postman发送消息

3.3.1. 使用Postman发送消息配图

3.3.1.1. 服务器订阅消息

3.3.1.1. 服务器订阅消息配图

3.3.1.2. MQTTBOX订阅消息

3.3.1.2. MQTTBOX订阅消息配图

3.3.2. MQTTBOX发送消息

3.3.2.1. MQTTBOX订阅消息
3.3.2.2. 代码订阅消息

3.3.2.2. 代码订阅消息配图

原文链接: https://juejin.cn/post/7362883850240901120

文章标签