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 协议开启成功，到这中间件环境就搭建完毕了。
image.png
使用MQTT 协议默认的交换机 Exchange 为 amp.topic，而我们订阅的主题会在 Queues 注册一个客户端队列，路由 Routing key 就是我们设置的主题。
image.png
服务端消息发送
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
image.png
再看一下前端订阅消息的效果，看到消息被实时推送到了前端，这里只做了未读消息数量统计，一般还会做未读消息详情列表。
image.png
总结
未读消息是一个十分常见的功能，不管是 web端还是移动端系统都是必备的模块，MQTT 协议只是其中的一种实现方式，还是有必要掌握一种方法。具体用什么工具实现还是要看具体的业务场景和学习成本，像我用RabbitMQ 做还考虑到一些运维成本在里边。
网友评论