springboot+rabbitmq搭建实现mqtt协议订阅发布

古城微笑少年丶 2023-10-04 19:58 19阅读 0赞

`rabbitmq`大家熟知的是作为消息队列使用。而`rabbitmq`有两种协议，我们平时接触的消息队列是用的`AMQP`协议，而用在智能硬件中的是`MQTT`协议。

>     mqtt是一种基于发布/订阅（publish/subscribe）模式的轻量级通讯协议，通过订阅相应的主题来获取消息，是物联网（Internet of Thing）中的一个标准传输协议。

而本篇文章讲的是mqtt的简单实现（http://mp.weixin.qq.com/s/udFE6k9pPetIWsa6KeErrA）。

#### **1.首先搭建一下rabbitmq服务，简单展示一下安装所需要的命令** ####

执行安装命令，启动/暂停rabbitmq

brew install rabbitmq//安装
    rabbitmq-server//启动
    rabbitmqctl stop//停止
    rabbitmqctl status//状态
    brew services restart rabbitmq//服务升级后重新启动

登陆rabbitmq管理端，管理端的默认用户名、密码均为guest。访问http://localhost:15672/

#### **2.启用 rabbitmq的mqtt协议** ####

我们先开启 `rabbitmq` 的 `mqtt`协议，因为默认安装下是关闭的，命令如下：

rabbitmq-plugins enable rabbitmq_mqtt

#### **3.代码实现** ####

先在pom中添加依赖包

<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>1.2.0</version>
            </dependency>

application.yml的所需的配置

mqtt:
        config:
            username: guest
            password: guest
            hostUrl: tcp://127.0.0.1:1883
            clientId: ${random.value}
            pubTopic: pubTopic
            subTopic: subTopic

接下来是mqtt配置文件bean类：

import lombok.Data;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.stereotype.Component;
    
    
    /**
     * mqtt配置类
     *
     * @author WeiWei
     * @date 2022/08/08
     */
    @Data
    @ConfigurationProperties(prefix = "mqtt.config")
    @Component
    public class MqttConfig {
    
        /**
         * 用户名
         */
        private String username;
    
        /**
         * 密码
         */
        private String password;
    
        /**
         * 主机url(代理端即 broker)
         */
        private String hostUrl;
    
        /**
         * 客户机id
         */
        private String clientId;
    
        /**
         * 发布的主题
         */
        private String pubTopic;
    
        /**
         * 订阅的主题
         */
        private String subTopic;
    
    
    }

然后是mqtt配置发布，订阅的配置文件

import lombok.extern.slf4j.Slf4j;
    import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    
    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.outbound.MqttPahoMessageHandler;
    import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.MessageHandler;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.Resource;
    import java.util.Objects;
    
    /**
     * mqtt服务类
     * 一种基于发布/订阅（publish/subscribe）模式的轻量级通讯协议，通过订阅相应的主题来获取消息，
     * 是物联网（Internet of Thing）中的一个标准传输协议
     * ClientId是MQTT客户端的标识。MQTT服务端用该标识来识别客户端。因此ClientId必须是独立的。
     * clientID需为全局唯一。如果不同的设备使用相同的clientID同时连接物联网平台，那么先连接的那个设备会被强制断开。
     * @author WeiWei
     * @date 2022/08/08
     */
    @Slf4j
    @Configuration
    public class MqttServer{
    
        /**
         * 出站通道
         */
        public static final String OUTBOUND_CHANNEL = "mqttOutboundChannel";
    
        /**
         * 输入通道
         */
        public static final String INPUT_CHANNEL = "mqttInputChannel";
    
        /**
         * mqtt配置
         */
        @Resource
        private MqttConfig mqttConfig;
    
    
        /**
         * 初始化
         */
        @PostConstruct
        public void init() {
            log.info(mqttConfig.toString());
        }
    
        /**
         * mqtt客户工厂
         *
         * @return {@link MqttPahoClientFactory}
         */
        @Bean
        public MqttPahoClientFactory clientFactory() {
    
            MqttConnectOptions options = new MqttConnectOptions();
            //配置MqttConnectOptions
            options.setServerURIs(new String[]{mqttConfig.getHostUrl()});
            options.setUserName(mqttConfig.getUsername());
            options.setPassword(mqttConfig.getPassword().toCharArray());
            DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
            factory.setConnectionOptions(options);
            return factory;
        }
    
        /**
         * mqtt出站通道
         *
         * @return {@link MessageChannel}
         */
        @Bean(value = OUTBOUND_CHANNEL)
        public MessageChannel mqttOutboundChannel() {
            return new DirectChannel();
        }
    
        /**
         * mqtt出站handler
         *
         * @return {@link MessageHandler}
         */
        @Bean
        @ServiceActivator(inputChannel = OUTBOUND_CHANNEL)
        public MessageHandler mqttOutboundHandler() {
            //MqttPahoMessageHandler初始化
            MqttPahoMessageHandler handler = new MqttPahoMessageHandler(mqttConfig.getClientId(), clientFactory());
            //设置默认的qos级别
            handler.setDefaultQos(1);
            //保留标志的默认值。如果没有mqtt_retained找到标题，则使用它。如果提供了自定义，则不使用它converter。这里不启用
            handler.setDefaultRetained(false);
            //设置发布的主题
            handler.setDefaultTopic(mqttConfig.getPubTopic());
            //当 时true，调用者不会阻塞。相反，它在发送消息时等待传递确认。默认值为false（在确认交付之前发送阻止）。
            handler.setAsync(false);
            //当 async 和 async-events 都为 true 时，会发出 MqttMessageSentEvent（请参阅事件）。它包含消息、主题、客户端库生成的messageId、clientId和clientInstance（每次连接客户端时递增）。当客户端库确认交付时，会发出 MqttMessageDeliveredEvent。它包含 messageId、clientId 和 clientInstance，使传递与发送相关联。任何 ApplicationListener 或事件入站通道适配器都可以接收这些事件。请注意，有可能在 MqttMessageSentEvent 之前接收到 MqttMessageDeliveredEvent。默认值为false。
            handler.setAsyncEvents(false);
            return handler;
        }
    
        /**
         * mqtt输入通道
         *
         * @return {@link MessageChannel}
         */
        @Bean
        public MessageChannel mqttInputChannel() {
            return new DirectChannel();
        }
        
        /**
         * 入站
         *
         * @return {@link MessageProducer}
         */
        @Bean
        public MessageProducer inbound() {
            //配置订阅端MqttPahoMessageDrivenChannelAdapter
            MqttPahoMessageDrivenChannelAdapter adapter =
                    new MqttPahoMessageDrivenChannelAdapter(
                            mqttConfig.getClientId() + "_inbound", clientFactory(), mqttConfig.getSubTopic().split(","));
            //设置超时时间
            adapter.setCompletionTimeout(3000);
            //设置默认的消息转换类
            adapter.setConverter(new DefaultPahoMessageConverter());
            //设置qos级别
            adapter.setQos(1);
            //设置入站管道
            adapter.setOutputChannel(mqttInputChannel());
            return adapter;
        }
        
        /**
         * 消息处理程序
         *
         * @return {@link MessageHandler}
         */
        @Bean
        @ServiceActivator(inputChannel = INPUT_CHANNEL)
        public MessageHandler messageHandler() {
            return message -> {
                String topic = Objects.requireNonNull(message.getHeaders().get("mqtt_receivedTopic")).toString();
                log.info("订阅主题为: {}", topic);
                String[] topics = mqttConfig.getSubTopic().split(",");
                for (String t : topics) {
                    if (t.equals(topic)) {
                        log.info("接收到该主题消息为: {}", message.getPayload());
                    }
                }
            };
        }
    
    }

最后是外提供发送消息的`API`

import org.springframework.integration.annotation.MessagingGateway;
    import org.springframework.integration.mqtt.support.MqttHeaders;
    import org.springframework.messaging.handler.annotation.Header;
    import org.springframework.stereotype.Component;
    
    /**
     * mqtt网关（发布端需要用到）
     *
     * @author WeiWei
     * @date 2022/08/08
     */
    @Component
    @MessagingGateway(defaultRequestChannel = MqttServer.OUTBOUND_CHANNEL)
    public interface MqttGateway {
    
        /**
         * 发送到mqtt
         *
         * @param payload 有效载荷
         */
        void sendToMqtt(String payload);
    
        /**
         * 发送到mqtt
         *
         * @param topic   主题
         * @param payload 消息内容
         */
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
    
        /**
         * 发送到mqtt
         *
         * @param topic   主题
         * @param qos     qos
         * @param payload 消息内容
         */
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
    
    }

以上就完成了，可以利用mqtt测试工具测试。亲测有效