devbox
小蓝xlanll
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMQ实现即时通讯

作者：小蓝xlanll | 2024-04-25 08:25:13

赞

踩

RabbitMQ实现即时通讯

有时候我们的项目中会用到即时通讯功能，比如电商系统中的客服聊天功能，还有在支付过程中，当用户支付成功后，第三方支付服务会回调我们的回调接口，此时我们需要通知前端支付成功。最近发现RabbitMQ可以很方便的实现即时通讯功能，如果你没有特殊的业务需求，甚至可以不写后端代码，今天给大家讲讲如何使用RabbitMQ来实现即时通讯！

MQTT协议

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的轻量级通讯协议，该协议构建于TCP/IP协议上。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。

MQTT相关概念

  • Publisher（发布者）：消息的发出者，负责发送消息。

  • Subscriber（订阅者）：消息的订阅者，负责接收并处理消息。

  • Broker（代理）：消息代理，位于消息发布者和订阅者之间，各类支持MQTT协议的消息中间件都可以充当。

  • Topic（主题）：可以理解为消息队列中的路由，订阅者订阅了主题之后，就可以收到发送到该主题的消息。

  • Payload（负载）；可以理解为发送消息的内容。

  • QoS（消息质量）：全称Quality of Service，即消息的发送质量，主要有QoS 0、QoS 1、QoS 2三个等级，下面分别介绍下：

    • QoS 0（Almost Once）：至多一次，只发送一次，会发生消息丢失或重复；

    • QoS 1（Atleast Once）：至少一次，确保消息到达，但消息重复可能会发生；

    • QoS 2（Exactly Once）：只有一次，确保消息只到达一次。

RabbitMQ启用MQTT功能

RabbitMQ启用MQTT功能，需要先安装然RabbitMQ然后再启用MQTT插件。

rabbitmq-plugins enable rabbitmq_mqtt

  • 开启成功后，查看管理控制台，我们可以发现MQTT服务运行在1883端口上了。

MQTT客户端

我们可以使用MQTT客户端来测试MQTT的即时通讯功能，这里使用的是MQTTBox这个客户端工具。

  • 首先下载并安装好MQTTBox，下载地址：http://workswithweb.com/mqttbox.html

  • 点击Create MQTT Client按钮来创建一个MQTT客户端；

  • 接下来对MQTT客户端进行配置，主要是配置好协议端口、连接用户名密码和QoS即可；

  • 再配置一个订阅者，订阅者订阅testTopicA这个主题，我们会向这个主题发送消息；

  • 发布者向主题中发布消息，订阅者可以实时接收到。

前端直接实现即时通讯

既然MQTTBox客户端可以直接通过RabbitMQ实现即时通讯，那我们是不是直接使用前端技术也可以实现即时通讯？答案是肯定的！下面我们将通过html+javascript实现一个简单的聊天功能，真正不写一行后端代码实现即时通讯！

  • 由于RabbitMQ与Web端交互底层使用的是WebSocket，所以我们需要开启RabbitMQ的MQTT WEB支持，使用如下命令开启即可；

rabbitmq-plugins enable rabbitmq_web_mqtt

  • 开启成功后，查看管理控制台，我们可以发现MQTT的WEB服务运行在15675端口上了；

  • WEB端与MQTT服务进行通讯需要使用一个叫MQTT.js的库，项目地址：https://github.com/mqttjs/MQTT.js

  • 实现的功能非常简单，一个单聊功能，需要注意的是配置好MQTT服务的访问地址为：ws://localhost:15675/ws

  1. <!DOCTYPE html>
  2. <html lang="en">
  3. <head>
  4.     <meta charset="UTF-8">
  5.     <title>Title</title>
  6. </head>
  7. <body>
  8. <div>
  9.     <label>目标Topic：<input id="targetTopicInput" type="text"></label><br>
  10.     <label>发送消息：<input id="messageInput" type="text"></label><br>
  11.     <button onclick="sendMessage()">发送</button>
  12.     <button onclick="clearMessage()">清空</button>
  13.     <div id="messageDiv"></div>
  14. </div>
  15. </body>
  16. <script src="https://unpkg.com/mqtt/dist/mqtt.min.js"></script>
  17. <script>
  18.     //RabbitMQ的web-mqtt连接地址
  19.     const url = 'ws://localhost:15675/ws';
  20.     //获取订阅的topic
  21.     const topic = getQueryString("topic");
  22.     //连接到消息队列
  23.     let client = mqtt.connect(url);
  24.     client.on('connect', function () {
  25.         //连接成功后订阅topic
  26.         client.subscribe(topic, function (err) {
  27.             if (!err) {
  28.                 showMessage("订阅topic：" + topic + "成功！");
  29.             }
  30.         });
  31.     });
  32.     //获取订阅topic中的消息
  33.     client.on('message', function (topic, message) {
  34.         showMessage("收到消息：" + message.toString());
  35.     });
  36.  
  37.     //发送消息
  38.     function sendMessage() {
  39.         let targetTopic = document.getElementById("targetTopicInput").value;
  40.         let message = document.getElementById("messageInput").value;
  41.         //向目标topic中发送消息
  42.         client.publish(targetTopic, message);
  43.         showMessage("发送消息给" + targetTopic + "的消息：" + message);
  44.     }
  45.  
  46.     //从URL中获取参数
  47.     function getQueryString(name) {
  48.         let reg = new RegExp("(^|&)" + name + "=([^&]*)(&|$)", "i");
  49.         let r = window.location.search.substr(1).match(reg);
  50.         if (r != null) {
  51.             return decodeURIComponent(r[2]);
  52.         }
  53.         return null;
  54.     }
  55.  
  56.     //在消息列表中展示消息
  57.     function showMessage(message) {
  58.         let messageDiv = document.getElementById("messageDiv");
  59.         let messageEle = document.createElement("div");
  60.         messageEle.innerText = message;
  61.         messageDiv.appendChild(messageEle);
  62.     }
  63.  
  64.     //清空消息列表
  65.     function clearMessage() {
  66.         let messageDiv = document.getElementById("messageDiv");
  67.         messageDiv.innerHTML = "";
  68.     }
  69. </script>
  70. </html>

  • 接下来我们订阅不同的主题开启两个页面测试下功能（页面放在了SpringBoot应用的resource目录下了，需要先启动应用再访问）：

    • 第一个订阅主题testTopicA，访问地址：http://localhost:8088/page/index?topic=testTopicA

    • 第二个订阅主题testTopicB，访问地址：http://localhost:8088/page/index?topic=testTopicB

  • 之后互相发送消息，让我们来看看效果吧！

在SpringBoot中使用

没有特殊业务需求的时候，前端可以直接和RabbitMQ对接实现即时通讯。但是有时候我们需要通过服务端去通知前端，此时就需要在应用中集成MQTT了，接下来我们来讲讲如何在SpringBoot应用中使用MQTT。

  • 首先我们需要在pom.xml中添加MQTT相关依赖；

  1. <!--Spring集成MQTT-->
  2. <dependency>
  3.     <groupId>org.springframework.integration</groupId>
  4.     <artifactId>spring-integration-mqtt</artifactId>
  5. </dependency>

  • 在application.yml中添加MQTT相关配置，主要是访问地址、用户名密码、默认主题信息；

  1. rabbitmq:
  2.   mqtt:
  3.     url: tcp://localhost:1883
  4.     username: guest
  5.     password: guest
  6.     defaultTopic: testTopic

  • 编写一个Java配置类从配置文件中读取配置便于使用；

  1. /**
  2.  * MQTT相关配置
  3.  * Created by macro on 2020/9/15.
  4.  */
  5. @Data
  6. @EqualsAndHashCode(callSuper = false)
  7. @Component
  8. @ConfigurationProperties(prefix = "rabbitmq.mqtt")
  9. public class MqttConfig {
  10.     /**
  11.      * RabbitMQ连接用户名
  12.      */
  13.     private String username;
  14.     /**
  15.      * RabbitMQ连接密码
  16.      */
  17.     private String password;
  18.     /**
  19.      * RabbitMQ的MQTT默认topic
  20.      */
  21.     private String defaultTopic;
  22.     /**
  23.      * RabbitMQ的MQTT连接地址
  24.      */
  25.     private String url;
  26. }

  • 添加MQTT消息订阅者相关配置，使用@ServiceActivator注解声明一个服务激活器，通过MessageHandler来处理订阅消息；

  1. /**
  2.  * MQTT消息订阅者相关配置
  3.  * Created by macro on 2020/9/15.
  4.  */
  5. @Slf4j
  6. @Configuration
  7. public class MqttInboundConfig {
  8.     @Autowired
  9.     private MqttConfig mqttConfig;
  10.  
  11.     @Bean
  12.     public MessageChannel mqttInputChannel() {
  13.         return new DirectChannel();
  14.     }
  15.  
  16.     @Bean
  17.     public MessageProducer inbound() {
  18.         MqttPahoMessageDrivenChannelAdapter adapter =
  19.                 new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getUrl(), "subscriberClient",
  20.                         mqttConfig.getDefaultTopic());
  21.         adapter.setCompletionTimeout(5000);
  22.         adapter.setConverter(new DefaultPahoMessageConverter());
  23.         //设置消息质量：0->至多一次；1->至少一次；2->只有一次
  24.         adapter.setQos(1);
  25.         adapter.setOutputChannel(mqttInputChannel());
  26.         return adapter;
  27.     }
  28.  
  29.     @Bean
  30.     @ServiceActivator(inputChannel = "mqttInputChannel")
  31.     public MessageHandler handler() {
  32.         return new MessageHandler() {
  33.  
  34.             @Override
  35.             public void handleMessage(Message<?> message) throws MessagingException {
  36.                 //处理订阅消息
  37.                 log.info("handleMessage : {}",message.getPayload());
  38.             }
  39.  
  40.         };
  41.     }
  42. }

  • 添加MQTT消息发布者相关配置；

  1. /**
  2.  * MQTT消息发布者相关配置
  3.  * Created by macro on 2020/9/15.
  4.  */
  5. @Configuration
  6. public class MqttOutboundConfig {
  7.  
  8.     @Autowired
  9.     private MqttConfig mqttConfig;
  10.  
  11.     @Bean
  12.     public MqttPahoClientFactory mqttClientFactory() {
  13.         DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
  14.         MqttConnectOptions options = new MqttConnectOptions();
  15.         options.setServerURIs(new String[] { mqttConfig.getUrl()});
  16.         options.setUserName(mqttConfig.getUsername());
  17.         options.setPassword(mqttConfig.getPassword().toCharArray());
  18.         factory.setConnectionOptions(options);
  19.         return factory;
  20.     }
  21.  
  22.     @Bean
  23.     @ServiceActivator(inputChannel = "mqttOutboundChannel")
  24.     public MessageHandler mqttOutbound() {
  25.         MqttPahoMessageHandler messageHandler =
  26.                 new MqttPahoMessageHandler("publisherClient", mqttClientFactory());
  27.         messageHandler.setAsync(true);
  28.         messageHandler.setDefaultTopic(mqttConfig.getDefaultTopic());
  29.         return messageHandler;
  30.     }
  31.  
  32.     @Bean
  33.     public MessageChannel mqttOutboundChannel() {
  34.         return new DirectChannel();
  35.     }
  36. }

  • 添加MQTT网关，用于向主题中发送消息；

  1. /**
  2.  * MQTT网关，通过接口将数据传递到集成流
  3.  * Created by macro on 2020/9/15.
  4.  */
  5. @Component
  6. @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
  7. public interface MqttGateway {
  8.     /**
  9.      * 发送消息到默认topic
  10.      */
  11.     void sendToMqtt(String payload);
  12.  
  13.     /**
  14.      * 发送消息到指定topic
  15.      */
  16.     void sendToMqtt(String payload, @Header(MqttHeaders.TOPIC) String topic);
  17.  
  18.     /**
  19.      * 发送消息到指定topic并设置QOS
  20.      */
  21.     void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
  22. }

  • 添加MQTT测试接口，使用MQTT网关向特定主题中发送消息；

  1. /**
  2.  * MQTT测试接口
  3.  * Created by macro on 2020/9/15.
  4.  */
  5. @Api(tags = "MqttController", description = "MQTT测试接口")
  6. @RestController
  7. @RequestMapping("/mqtt")
  8. public class MqttController {
  9.  
  10.     @Autowired
  11.     private MqttGateway mqttGateway;
  12.  
  13.     @PostMapping("/sendToDefaultTopic")
  14.     @ApiOperation("向默认主题发送消息")
  15.     public CommonResult sendToDefaultTopic(String payload) {
  16.         mqttGateway.sendToMqtt(payload);
  17.         return CommonResult.success(null);
  18.     }
  19.  
  20.     @PostMapping("/sendToTopic")
  21.     @ApiOperation("向指定主题发送消息")
  22.     public CommonResult sendToTopic(String payload, String topic) {
  23.         mqttGateway.sendToMqtt(payload, topic);
  24.         return CommonResult.success(null);
  25.     }
  26. }

  • 调用接口向主题中发送消息进行测试；

  • 后台成功接收到消息并进行打印。

  1. 2020-09-17 14:29:01.689  INFO 11192 --- [ubscriberClient] c.m.mall.tiny.config.MqttInboundConfig   : handleMessage : 来自网页上的消息
  2. 2020-09-17 14:29:06.101  INFO 11192 --- [ubscriberClient] c.m.mall.tiny.config.MqttInboundConfig   : handleMessage : 来自网页上的消息
  3. 2020-09-17 14:29:07.384  INFO 11192 --- [ubscriberClient] c.m.mall.tiny.config.MqttInboundConfig   : handleMessage : 来自网页上的消息

总结

消息中间件应用越来越广泛，不仅可以实现可靠的异步通信，还可以实现即时通讯，掌握一个消息中间件还是很有必要的。如果没有特殊业务需求，客户端或者前端直接使用MQTT对接消息中间件即可实现即时通讯，有特殊需求的时候也可以使用SpringBoot集成MQTT的方式来实现，总之消息中间件是实现即时通讯的一个好选择！

项目源码地址

https://github.com/macrozheng/mall-learning/tree/master/mall-tiny-mqtt

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小蓝xlanll/article/detail/484227?site
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号