在完成了基于AMQP协议的RabbitMQ消息收发后，我们要继续实现基于MQTT协议的RabbitMQ消息收发。
由于C#的RabbitMQ.Client包中只实现了基于AMQP协议的消息收发功能的封装，所以要实现基于MQTT协议的收发，我们要下载新的包。
在NuGet的解决方案中，我们选择了简单实用的M2Mqtt。
关于M2Mqtt的资料，可以参考： https://m2mqtt.wordpress.com/ https://github.com/eclipse/paho.mqtt.m2mqtt
消费者代码：
using System;
using uPLibrary.Networking.M2Mqtt;
using uPLibrary.Networking.M2Mqtt.Messages; namespace MQTTDemo
{
class Client
{
static void Main()
{
// create client instance
MqttClient client = new MqttClient("127.0.0.1"); // register to message received
client.MqttMsgPublishReceived += client_MqttMsgPublishReceived; string clientId = Guid.NewGuid().ToString();
client.Connect(clientId); client.Subscribe(new string[] { "test" }, new byte[] { MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE });
} static void client_MqttMsgPublishReceived(object sender, MqttMsgPublishEventArgs e)
{
string msg = System.Text.Encoding.Default.GetString(e.Message);
Console.WriteLine(msg);
}
}
}
生产者代码：
using System;
using System.Text;
using uPLibrary.Networking.M2Mqtt;
using uPLibrary.Networking.M2Mqtt.Messages; namespace MQTTServer
{
class Server
{
static void Main()
{
// create client instance
MqttClient client = new MqttClient("127.0.0.1"); string clientId = Guid.NewGuid().ToString();
client.Connect(clientId); client.Publish("test", Encoding.UTF8.GetBytes("hello"), MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE, false); Console.WriteLine("Publish!!!"); Console.ReadKey();
client.Disconnect();
}
}
}
消费者监听的队列名会基于产生的Guid进行前后封装，“test”表示的是topic值，选择QOS_LEVEL_AT_MOST_ONCE而不是QOS_LEVEL_EXACTLY_ONCE是因为测试发现QOS_LEVEL_EXACTLY_ONCE消息会被收到多次（我也不知道为啥）。
消费者监听的队列会在消费者程序结束后自动删除，生产者不产生队列。
在rabbitmq-plugins enable rabbitmq_mqtt之后，我们就可以愉快地通过MQTT收发消息了。
然而，我们发现只能通过127.0.0.1和localhost访问RabbitMQ服务器，而本机IP访问失败。
查阅了大量资料后，我发现这是由于rabbitmq默认的config中有这么一段文字，所以我们之能在localhost中访问服务器。
%% The default "guest" user is only permitted to access the server
%% via a loopback interface (e.g. localhost).
%% {loopback_users, [<<"guest">>]},
所以我们取消了{loopback_users, []}的注释
%% Uncomment the following line if you want to allow access to the
%% guest user from anywhere on the network.
%% {loopback_users, []},
值得注意的是，由于我们在config中仅仅取消了一行注释，所以这段代码是整个代码块的最后一行。于是我们应该将句末的逗号一同去掉。
然而，我发现怎么更改默认启动的rabbitmq对应的comfig文件，都无法成功地使用我更改后的config文件，察看了log发现用的是不存在的rabbitmq.conf文件。
修改成rabbitmq.conf后服务启动失败，所以我放弃了直接在默认启动服务中更改。
由于之前配置过rabbitmq集群，所以我打算采用集群的方式解决问题。
操作可以参考https://www.cnblogs.com/lucifer1997/p/9324130.html，其中我将ClusterNode1改为了mqtt，同时在rabbitmq-mqtt.config中对{loopback_users, []}进行了更改。
如果要修改默认的mqtt用户、密码、虚拟用户、交换机信息，可以参照http://www.rabbitmq.com/mqtt.html在rabbitmq-mqtt.config中进行修改。
在命令行操作之前先把原来开启的rabbitmq_mqtt停用，避免两个服务同时监听1883端口导致报错。 rabbitmq-plugins disable rabbitmq_mqtt
同时在操作了rabbitmq-plugins-mqtt enable rabbitmq_management之后执行rabbitmq-plugins-mqtt enable rabbitmq_mqtt。
如此就可以在集群后实现远程MQTT收发，同时还可以实现AMQP与MQTT之间的收发。