Mqtt实践-未读消息（小红点），前端与 RabbitMQ 实时消息推送实践

喜欢ヅ旅行 2021-06-10 20:40 769阅读 0赞

`web` 端实时消息推送，常用的实现方式比较多，但万变不离其宗，底层基本上还是依赖于 `websocket`，`MQTT` 协议也不例外。

## RabbitMQ 搭建 ##

`RabbitMQ `的基础搭建就不详细说了，自行百度一步一步搞问题不大，这里主要说一下两个比较重要的配置。

### 1、开启 mqtt 协议 ###

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

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

rabbitmq-plugins enable rabbitmq_mqtt

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

rabbitmq-plugins enable rabbitmq_web_mqtt

在 `RabbitMQ` 管理后台看到如下的显示，就表示`MQTT` 协议开启成功，到这中间件环境就搭建完毕了。

![协议对应端口号][20200908110952935.png]

使用`MQTT` 协议默认的交换机 `Exchange` 为 `amp.topic`，而我们订阅的主题会在 `Queues` 注册一个客户端队列，路由 `Routing key` 就是我们设置的主题。

![交换机信息][20200908133812233.png]

## 服务端消息发送 ##

`web` 端实时消息推送一般都是单向的推送，前端接收服务端推送的消息显示即可，所以就只实现消息发送即可。

### 1、mqtt 客户端依赖包 ###

引入 `spring-integration-mqtt`、`org.eclipse.paho.client.mqttv3` 两个工具包实现

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

### 2、消息发送者 ###

消息的发送比较简单，主要是应用到 `@ServiceActivator` 注解，需要注意`messageHandler.setAsync`属性，如果设置成 `false`，关闭异步模式发送消息时可能会阻塞。

@Configuration
    public class IotMqttProducerConfig {
    
        @Autowired
        private MqttConfig mqttConfig;
    
        @Bean
        public MqttPahoClientFactory mqttClientFactory() {
            DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
            factory.setServerURIs(mqttConfig.getServers());
            return factory;
        }
    
        @Bean
        public MessageChannel mqttOutboundChannel() {
            return new DirectChannel();
        }
    
        @Bean
        @ServiceActivator(inputChannel = "iotMqttInputChannel")
        public MessageHandler mqttOutbound() {
            MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfig.getServerClientId(), mqttClientFactory());
            messageHandler.setAsync(false);
            messageHandler.setDefaultTopic(mqttConfig.getDefaultTopic());
            return messageHandler;
        }
    }

`MQTT` 对外提供发送消息的 `API` 时，需要使用 `@MessagingGateway` 注解，去提供一个消息网关代理，参数 `defaultRequestChannel` 指定发送消息绑定的`channel`。

可以实现三种`API`接口，`payload` 为发送的消息，`topic` 发送消息的主题，`qos` 消息质量。

@MessagingGateway(defaultRequestChannel = "iotMqttInputChannel")
    public interface IotMqttGateway {
    
        // 向默认的 topic 发送消息
        void sendMessage2Mqtt(String payload);
        // 向指定的 topic 发送消息
        void sendMessage2Mqtt(String payload,@Header(MqttHeaders.TOPIC) String topic);
        // 向指定的 topic 发送消息，并指定服务质量参数
        void sendMessage2Mqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
    }

## 前端消息订阅 ##

前端使用与服务端对应的工具 `paho-mqtt` `mqttws31.js`实现，实现方式与传统的 `websocket` 方式差不多，核心方法 `client = new Paho.MQTT.Client` 和 各种监听事件，代码比较简洁。

**注意**：要保证前后端 `clientId`的全局唯一性，我这里就简单用随机数解决了

## 测试 ##

前后端的代码并不多，接下来我们测试一下，弄了个页面看看效果。

首先用 `postman` 模拟后端发送消息

http://127.0.0.1:8080/fun/sendMessage?message=我是程序员内点事&topic=push_message_topic

![模拟发送消息][20200908132348293.png]

再看一下前端订阅消息的效果，看到消息被实时推送到了前端，这里只做了未读消息数量统计，一般还会做未读消息详情列表。

![实时消息推送动图][20200908132757419.gif_pic_center]

## 总结 ##

未读消息是一个十分常见的功能，不管是 `web`端还是移动端系统都是必备的模块，`MQTT` 协议只是其中的一种实现方式，还是有必要掌握一种方法。具体用什么工具实现还是要看具体的业务场景和学习成本，像我用`RabbitMQ` 做还考虑到一些运维成本在里边。

**本文完整代码地址**：[https://github.com/chengxy-nds/Springboot-Notebook][https_github.com_chengxy-nds_Springboot-Notebook]

[20200908110952935.png]: /images/20210527/8b1e0832a1814ce2be8a390de1adaff6.png
[20200908133812233.png]: /images/20210527/1fcf1036fe354e19bde0baff07737581.png
[20200908132348293.png]: /images/20210527/72c04b3dfacb4e7b9eea61f60f692a49.png
[20200908132757419.gif_pic_center]: /images/20210527/2d9454355d4c49cba25371655fc4a481.png
[https_github.com_chengxy-nds_Springboot-Notebook]: https://github.com/chengxy-nds/Springboot-Notebook