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

rabbitmq-plugins enable rabbitmq_mqtt

rabbitmq-plugins enable rabbitmq_web_mqtt 

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

@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;
    }
}

@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);
}

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

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