前端与RabbitMQ实时消息推送未读消息小红点实现示例

目录

引言

RabbitMQ 搭建

1、开启 mqtt 协议

服务端消息发送

1、mqtt 客户端依赖包

2、消息发送者

前端消息订阅

测试

总结

引言

前几天粉丝群里有个小伙伴问过:web 页面的未读消息(小红点)怎么实现比较简单,刚好本周手头有类似的开发任务,索性就整理出来供小伙伴们参考,没准哪天就能用得上呢。

之前在 《springboot + rabbitmq 做智能家居》 中说过可以用 rabbitmq 的 MQTT 协议做智能家居的指令推送,里边还提到过能用 MQTT 协议做 web 的消息推送,而未读消息(小红点)功能刚好应用到实时消息推送了。

MQTT 协议就不再赘述了,没接触过的同学翻翻前边的文章温习一下吧,今天还是主要以实践为主!

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

RabbitMQ 搭建

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

1、开启 mqtt 协议

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

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

rabbitmq-plugins enable rabbitmq_mqtt

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

rabbitmq-plugins enable rabbitmq_web_mqtt 

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

协议对应端口号

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

交换机信息

服务端消息发送

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

1、mqtt 客户端依赖包

引入 spring-integration-mqttorg.eclipse.paho.client.mqttv3 两个工具包实现

<!--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>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的全局唯一性,我这里就简单用随机数解决了

<script type="text/javascript">         // mqtt协议rabbitmq服务         var brokerIp = location.hostname;         // mqtt协议端口号         var port = 15675;         // 接受推送消息的主题         var topic = "push_message_topic";         // mqtt连接         client = new Paho.MQTT.Client(brokerIp, port, "/ws", "clientId_" + parseInt(Math.random() * 100, 10));         var options = {             timeout: 3, //超时时间             keepAliveInterval: 30,//心跳时间             onSuccess: function () {                 console.log(("连接成功~"));                 client.subscribe(topic, {qos: 1});             },             onFailure: function (message) {                 console.log(("连接失败~" + message.errorMessage));             }         };         // 考虑到https的情况         if (location.protocol == "https:") {             options.useSSL = true;         }         client.connect(options);         console.log(("已经连接到" + brokerIp + ":" + port));         // 连接断开事件         client.onConnectionLost = function (responseObject) {             console.log("失去连接 - " + responseObject.errorMessage);         };         // 接收消息事件         client.onMessageArrived = function (message) {             console.log("接受主题: " + message.destinationName + "的消息: " + message.payloadString);             $("#arrivedDiv").append("<br/>"+message.payloadString);             var count = $("#count").text();             count = Number(count) + 1;             $("#count").text(count);         };         // 推送给指定主题         function sendMessage() {             var a = $("#message").val();             if (client.isConnected()) {                 var message = new Paho.MQTT.Message(a);                 message.destinationName = topic;                 client.send(message);             }         }     </script> 测试

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

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

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

模拟发送消息

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

实时消息推送动图

总结

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

本文完整代码地址:https://github.com/chengxy-nds/Springboot-Notebook/tree/master/springboot-mqtt-messagepush

以上就是前端与RabbitMQ实时消息推送未读消息小红点实现示例的详细内容,更多关于前端 RabbitMQ实时消息推送的资料请关注易知道(ezd.cc)其它相关文章!

推荐阅读