RabbitMQ之Plugins插件----AMQP对接MQTT

本文主要是介绍RabbitMQ之Plugins插件----AMQP对接MQTT，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1.启用插件

rabbitmq-plugins enable rabbitmq_mqtt

2.检查是否启动成功，打开rabbitmq后台

3.概念：
AMQP是由交换器和queue队列组成的消息队列机制，MQTT是由订阅主题组成的消息机制

1.MQTT创建连接时会向rabbitmq创建一个自己的queue，默认使用amq.topic交换器，由mqtt-subscription-***-qos1组成queue名。其中qos概念自行了解。

2.AMQP连接消费也需要绑定到MQTT对应的交换器和队列中，其中routingKey则使用MQTT订阅主题。

上代码：

//****************************************************************** */
//MQTTconst mqtt = require("mqtt")
const client = mqtt.connect({url: "mqtt://127.0.0.1:1883",clientId: 'nickchen111', // 客户端ID，此ID可自定义。queue后缀username: 'guest',   // 用户名password: 'guest'
}) // 连接到mqtt服务器var quque = "device_aaa"
var quque_rpc = "device_aaa_rpc";
client.subscribe(quque_rpc, { qos: 1 })var status = false;
var count = 0;
var error = 0;
setInterval(() => {if (status) {count += 1;var msg = Math.ceil(Math.random() * 40)console.log("MQTT发送消息:" + msg)client.publish(quque, msg.toString(), { qos: 0, retain: true })}
}, 3000);client.on("message", function (top, message) {if (top == "device_aaa") {error += 1;}console.log("MQTT接收消息:" + top + "  : " + message.toString() + "*******丢失返回消息共:[" + error + "]条")if (JSON.parse(message.toString()).command == "stop") {status = false;}if (JSON.parse(message.toString()).command == "start") {status = true;}
})//****************************************************************** */
//AMQP
const amqp = require('amqplib');async function consumer() {// 创建链接对象const connection = await amqp.connect('amqp://localhost:5672');// 获取通道const channel = await connection.createChannel();// 声明参数const exchangeName = 'amq.topic';const queueName = 'mqtt-subscription-nickchen111qos1';const routingKey = 'device_aaa';// // 声明一个交换机// await channel.assertExchange(exchangeName, 'amq.topic', { durable: true });//消息持久化// // 声明一个队列await channel.assertQueue(queueName, { autoDelete: true, durable: true });// 绑定关系（队列、交换机、路由键）await channel.bindQueue(queueName, exchangeName, routingKey);var data = await channel.publish(exchangeName, "device_aaa_rpc", Buffer.from(JSON.stringify({ command: "start" })))// 消费await channel.consume(queueName, msg => {console.log('Consumer：', msg.content.toString());channel.ack(msg);},{ noAck: false });//开启通知console.log('消费端启动成功！');
}
consumer();