前端与RabbitMQ实时消息推送怎么实现

发布时间：2022-02-25 10:53:29 作者：iii
来源：亿速云阅读：276

这篇文章主要讲解了“前端与RabbitMQ实时消息推送怎么实现”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“前端与RabbitMQ实时消息推送怎么实现”吧！

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 协议开启成功，到这中间件环境就搭建完毕了。

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

服务端消息发送

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

1、mqtt 客户端依赖包

引入 spring-integration-mqtt、org.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

感谢各位的阅读，以上就是“前端与RabbitMQ实时消息推送怎么实现”的内容了，经过本文的学习后，相信大家对前端与RabbitMQ实时消息推送怎么实现这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是亿速云，小编将为大家推送更多相关知识点的文章，欢迎关注！