zz22 发表于 2017-12-24 06:03:21

快速搭建MQTT服务器(MQTTnet和Apache Apollo)

前言
  MQTT协议是IBM开发的一个即时通讯协议,有可能成为物联网的重要组成部分,http://mqtt.org/。

  MQTT is a machine-to-machine (M2M)/"Internet of Things" connectivity protocol. It was designed as an extremely lightweight publish/subscribe messaging transport. It is useful for connections with remote locations where a small code footprint is required and/or network bandwidth is at a premium. For example, it has been used in sensors communicating to a broker via satellite link, over occasional dial-up connections with healthcare providers, and in a range of home automation and small device scenarios. It is also>
  通过https://github.com/mqtt/mqtt.github.io/wiki/servers 找到官方推荐的服务端软件,比如:Apache Apollo,
  通过https://github.com/mqtt/mqtt.github.io/wiki/libraries可以找到推荐的客户端类库,比如:Eclipse Paho Java

MQTTnet
  MQTTnet 是MQTT协议的.NET 开源类库。

  MQTTnet is a .NET library for MQTT based communication. It provides a MQTT client and a MQTT server. The implementation is based on the documentation from http://mqtt.org/.

  通过Nuget搜索MQTT找到了MQTTnet,它不是下载量最多的,也不在官方推荐列表中,主要是因为同时支持客户端和服务端,所以开始下载试用,结果证明有坑,源码在vs2015中不能打开,客户端示例接收不到消息。
  开源地址:https://github.com/chkr1011/MQTTnet
  首先把官方的控制台程序改成winform的,界面如下:



  

public partial>
{  

  

private MqttServer mqttServer = null;  

private MqttClient mqttClient = null;  

  

public Form1()  
{
  
InitializeComponent();
  
}
  

  

private void button_启动服务端_Click(object sender, EventArgs e)  
{
  

  
MqttTrace.TraceMessagePublished
+= MqttTrace_TraceMessagePublished;  

  

  

if (this.mqttServer == null)  
{
  

try  
{
  

var options = new MqttServerOptions  
{
  
ConnectionValidator
= p =>  
{
  

if (p.ClientId == "SpecialClient")  
{
  

if (p.Username != "USER" || p.Password != "PASS")  
{
  

return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;  
}
  
}
  

  

return MqttConnectReturnCode.ConnectionAccepted;  
}
  
};
  

  
mqttServer
= new MqttServerFactory().CreateMqttServer(options);  

  
}
  

catch (Exception ex)  
{
  
MessageBox.Show(ex.Message);
  

return;  

  
}
  
}
  
mqttServer.Start();
  

  

this.txt_服务器.AppendText( $">> 启动成功..." + Environment.NewLine);  
}
  

  

private void MqttTrace_TraceMessagePublished(object sender, MqttTraceMessagePublishedEventArgs e)  
{
  

this.Invoke(new Action(() =>  
{
  

this.txt_服务器.AppendText($">> [{e.ThreadId}] [{e.Source}] [{e.Level}]: {e.Message}" + Environment.NewLine);  

if (e.Exception != null)  
{
  

this.txt_服务器.AppendText( e.Exception + Environment.NewLine);  
}
  
}));
  
}
  

  

private void button_停止服务端_Click(object sender, EventArgs e)  
{
  

if (mqttServer != null)  
{
  
mqttServer.Stop();
  
}
  

this.txt_服务器.AppendText( $">> 停止成功" + Environment.NewLine);  
}
  

  

private async void button_启动客户端_Click(object sender, EventArgs e)  
{
  

if (this.mqttClient == null)  
{
  

var options = new MqttClientOptions  
{
  
Server
= "192.168.2.54",  
ClientId
= "zbl",  
CleanSession
= true  
};
  

this.mqttClient = new MqttClientFactory().CreateMqttClient(options);  

this.mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived;  

this.mqttClient.Connected += MqttClient_Connected;  

this.mqttClient.Disconnected += MqttClient_Disconnected;  

  

  
}
  

try  
{
  

await this.mqttClient.ConnectAsync();  
}
  

catch(Exception ex)  
{
  

this.txt_客户端.AppendText( $"### CONNECTING FAILED ###" + Environment.NewLine);  
}
  

  

  
}
  

  

private void MqttClient_Connected(object sender, EventArgs e)  
{
  

  

this.Invoke(new Action( async () =>  
{
  

  

this.txt_客户端.AppendText( $"### CONNECTED WITH SERVER ###" + Environment.NewLine);  

await this.mqttClient.SubscribeAsync(new List<TopicFilter>{  

new TopicFilter("#", MqttQualityOfServiceLevel.AtMostOnce)  
});
  

this.txt_客户端.AppendText($"### SUBSCRIBED###" + Environment.NewLine);  
}));
  
}
  

  

private void MqttClient_Disconnected(object sender, EventArgs e)  
{
  

this.Invoke(new Action(() =>  
{
  

this.txt_客户端.AppendText( $"### DISCONNECTED FROM SERVER ###" + Environment.NewLine);  

  
}));
  
}
  

  

private void MqttClient_ApplicationMessageReceived(object sender, MQTTnet.Core.MqttApplicationMessageReceivedEventArgs e)  
{
  

this.Invoke(new Action(() =>  
{
  

this.txt_客户端.AppendText( $">> Topic:{e.ApplicationMessage.Topic} Payload:{Encoding.UTF8.GetString(e.ApplicationMessage.Payload)} QoS:{e.ApplicationMessage.QualityOfServiceLevel}   Retain:{e.ApplicationMessage.Retain}" + Environment.NewLine);  

  
}));
  
}
  

  

private async void button_停止客户端_Click(object sender, EventArgs e)  
{
  

if (this.mqttClient != null)  
{
  

await this.mqttClient.DisconnectAsync();  
}
  

this.txt_客户端.AppendText( $">> 停止成功" + Environment.NewLine);  
}
  

  

private async void button_发送_Click(object sender, EventArgs e)  
{
  

  

//var options = new MqttClientOptions  

//{  

//    Server = "localhost"  

//};  

//var client = new MqttClientFactory().CreateMqttClient(options);  

  

//await client.ConnectAsync();  

  
var applicationMessage = new MqttApplicationMessage(this.txt_topic.Text,
  
Encoding.UTF8.GetBytes(this.txt_message.Text), MqttQualityOfServiceLevel.AtMostOnce, false);
  

  
await this.mqttClient.PublishAsync(applicationMessage);
  

  
}
  

  
private void button_清空服务端_Click(object sender, EventArgs e)
  
{
  
this.txt_服务器.Text = "";
  
}
  
}
  


Form1.cs  需要注意的是按照作者说明是
  

while (true)  
{
  
Console.ReadLine();
  

  
var applicationMessage = new MqttApplicationMessage(
  
"A/B/C",
  
Encoding.UTF8.GetBytes("Hello World"),
  
    MqttQualityOfServiceLevel.AtLeastOnce,
  
false
  
);
  

  
await client.PublishAsync(applicationMessage);
  
}
  

  

  客户端死活都收不到消息,改成 MqttQualityOfServiceLevel.AtMostOnce  就可以了,找问题时尝试下载源码调试因为vs2015打不开项目也折腾了一会。这个问题提交了issues,期待作者回复。

Apache Apollo
  1.下载Apollo服务器,我这里用的是Binaries for Windows。下载后解压到一个文件夹,注意路径不要包含中文,安装手册
  2.创建Broker Instance,命令行cd到bin目录,执行/bin/apollo create mybroker,执行后就会在bin目录下创建mybroker文件夹。
  3.运行Broker Instance,命令行cd到mybroker/bin目录,执行mybroker/bin/apollo-broker.cmd run
  4.Web Administrator,地址 http://127.0.0.1:61680/ or https://127.0.0.1:61681/,默认账号 admin,密码 password


参考


[*] MQTT协议的简单介绍和服务器的安装,
[*]编写和MQTT服务器通信的Android客户端程序,
[*] MQTT】在Windows下搭建MQTT服务器
  搭建了MQTT服务端之后需要在Esp8266模块和手机App中分别实现客户端功能,稍后待续。。。。
页: [1]
查看完整版本: 快速搭建MQTT服务器(MQTTnet和Apache Apollo)