RabbitMQ开启MQTT协议支持

1）RabbitMQ启用MQTT插件

1. root@mq:/# rabbitmq-plugins enable rabbitmq_mqtt
2. Enabling plugins on node rabbit@mq:
3. rabbitmq_mqtt
4. The following plugins have been configured:
5.   rabbitmq_management
6.   rabbitmq_management_agent
7.   rabbitmq_mqtt
8.   rabbitmq_web_dispatch
9. Applying plugin configuration to rabbit@mq...
10. The following plugins have been enabled:
11.   rabbitmq_mqtt
13. started 1 plugins.
14. root@mq:/# rabbitmq-plugins enable rabbitmq_web_mqtt
15. Enabling plugins on node rabbit@mq:
16. rabbitmq_web_mqtt
17. The following plugins have been configured:
18.   rabbitmq_management
19.   rabbitmq_management_agent
20.   rabbitmq_mqtt
21.   rabbitmq_web_dispatch
22.   rabbitmq_web_mqtt
23. Applying plugin configuration to rabbit@mq...
24. The following plugins have been enabled:
25.   rabbitmq_web_mqtt
27. started 1 plugins.
28. root@mq:/#

复制代码

2）RabbitMQ管理控制台查看
假如插件启动乐成，rabbitmq会打开1883和15675端口：

3）用MQTTX工具测试

4）用eclipse paho客户端测试
添加依赖

1. <dependency>
2.     <groupId>org.eclipse.paho</groupId>
3.     <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
4.     <version>1.2.5</version>
5. </dependency>

复制代码

收发消息测试

1. @RestController
2. public class DemoController {
4.     @GetMapping("/publish")
5.     public String publish() throws MqttException {
7.         MqttClientPersistence persistence = new MemoryPersistence();;//内存持久化
8.         MqttClient client = new MqttClient("tcp://192.168.137.138:1883", "abc", persistence);
10.         //连接选项中定义用户名密码和其它配置
11.         MqttConnectOptions options = new MqttConnectOptions();
12.         options.setCleanSession(true);//参数为true表示清除缓存，也就是非持久化订阅者，这个时候只要参数设为true，一定是非持久化订阅者。而参数设为false时，表示服务器保留客户端的连接记录
13.         options.setAutomaticReconnect(true);//是否自动重连
14.         options.setConnectionTimeout(30);//连接超时时间  秒
15.         options.setKeepAliveInterval(10);//连接保持检查周期  秒
16.         options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1); //版本
17.         options.setUserName("xjs1919");
18.         options.setPassword("123321".toCharArray());
19.         // client.setManualAcks(true);
20.         client.connect(options);//连接
21.         //订阅topic
22.         client.subscribe("demoTopic", 2);
23.         // 设置回调，将来收到消息的时候会被回调
24.         client.setCallback(new MqttCallbackExtended() {
25.             @Override
26.             public void connectComplete(boolean reconnect, String serverURI) {
27.                 System.out.println("连接完成");
28.             }
30.             @Override
31.             public void connectionLost(Throwable cause) {
32.                 System.out.println("连接丢失");
33.             }
35.             @Override
36.             public void messageArrived(String topic, MqttMessage message) throws Exception {
37.                 System.out.println("收到消息，topic："+topic + ", msg:" + new String(message.getPayload()));
38.                 //client.messageArrivedComplete(message.getId(),message.getQos());
39.             }
41.             @Override
42.             public void deliveryComplete(IMqttDeliveryToken token) {
43.                 // Qos0: 当消息发送出去就回调
44.                 // Qos1: 当发送者收到了puback的时候的回调
45.                 // Qos2: 当发送者收到了pubcomp的时候的回调
46.                 System.out.println("消息发送完成");
47.             }
48.         });
50.         client.publish("demoTopic", "hello,这是一个测试消息!".getBytes(), 2, false);
52.         return "ok";
53.     }
55. }

复制代码

参考文章：
https://www.cnblogs.com/motion/p/14974024.html
https://blog.csdn.net/u013615903/article/details/131395264

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。