mqttnet消息推送与接收[通俗易懂]

大家好，又见面了，我是你们的朋友全栈君。

创建windows服务网上有很多，不多述；

服务端做好后一定要写bat安装卸载文件

install.bat

@echo.请稍等，MqttNetServiceAddUserAndPassword服务安装启动中…………
@echo off
@title 安装windows服务：MqttNetServiceAddUserAndPassword
@sc create MqttNetServiceAddUserAndPassword binPath=”%~dp0\MqttNetServiceAddUserAndPassword.exe”
@sc config MqttNetServiceAddUserAndPassword start= auto
@sc start MqttNetServiceAddUserAndPassword
@echo.MqttNetServiceAddUserAndPassword启动完毕
pause

//binPath=”%~dp0\MqttNetServiceAddUserAndPassword.exe” 当前路径，也可指定

delete.bat

@echo.服务MqttNetServiceAddUserAndPassword卸载中……….
@echo off
@sc stop MqttNetServiceAddUserAndPassword
@sc delete MqttNetServiceAddUserAndPassword
@echo off
@echo.MqttNetServiceAddUserAndPassword卸载完毕
@pause

服务端：

using MQTTnet;
using MQTTnet.Protocol;
using MQTTnet.Server;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.ServiceProcess;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Timers;

namespace MqttNetServiceAddUserAndPassword
{

public partial class Service1 : ServiceBase
{

private readonly static object locker = new object();
private MqttServer mqttServer = null;
private System.Timers.Timer timer = null;

private GodSharp.Sockets.SocketServer socketService = null;

//此集合用于判断写入日志在一段时间内不重，以客户端id为依据，最多2000个清零；
private List<string> subClientIDs = new List<string>();
public Service1()
{

InitializeComponent();
//创建一个定时器，检查5s内有多少客户端接入并将相关信息记录到日志中
timer = new System.Timers.Timer();
timer.AutoReset = true;
timer.Enabled = true;
timer.Interval = 5000;
timer.Elapsed += new ElapsedEventHandler(GetSubClientSAndSetShow);

}

protected override void OnStart(string[] args)
{

//开启服务
//CreateMQTTServer();

Task.Run(CreateMQTTServer);

if (timer.Enabled == false)
{

timer.Enabled = true;
timer.Start();
}
//创建socket服务端
//CreateServerSocket();
// SocketServer.StartSocketService();
}

protected override void OnStop()
{

if (timer.Enabled == true)
{

timer.Enabled = false;
timer.Stop();
}
}
/// <summary>
/// 开启服务
/// </summary>
private async Task CreateMQTTServer()
{

if (mqttServer == null)
{

var optionsBuilder = new MqttServerOptionsBuilder();
optionsBuilder.WithConnectionValidator(c =>
{

if (c.ClientId.Length < 5 || !c.ClientId.StartsWith(“Eohi_”))
{

c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;
return;
}

if (c.Username != “user” || c.Password != “123456”)
{

c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
return;
}
c.ReturnCode = MqttConnectReturnCode.ConnectionAccepted;
});
//指定 ip地址，默认为本地，但此方法不能使用ipaddress报错，有哪位大神帮解答，感激。
//options.WithDefaultEndpointBoundIPAddress(IPAddress.Parse(“”))
//指定端口
optionsBuilder.WithDefaultEndpointPort(1884);
//连接记录数，默认一般为2000
//optionsBuilder.WithConnectionBacklog(2000);
mqttServer = new MqttFactory().CreateMqttServer() as MqttServer;
string msg = null;

//将发送的消息加到日志
mqttServer.ApplicationMessageReceived += (s, e) =>
{

msg = @”发送消息的客户端id:” + e.ClientId + “\r\n”
+ “发送时间：” + DateTime.Now + “\r\n”
+ “发送消息的主题：” + e.ApplicationMessage.Topic + “\r\n”
+ “发送的消息内容：” + Encoding.UTF8.GetString(e.ApplicationMessage.Payload ?? new byte[0]) + “\r\n”
+ “————————————————–\r\n”
;
WriteMsgLog(msg);
};
await mqttServer.StartAsync(optionsBuilder.Build());

}
}
#region 记录日志
/// <summary>
/// 消息记录日志
/// </summary>
/// <param name=”msg”></param>
private void WriteMsgLog(string msg)
{

//string path = @”C:\log.txt”;

//该日志文件会存在windows服务程序目录下
string path = AppDomain.CurrentDomain.BaseDirectory + “\\Msglog.txt”;
FileInfo file = new FileInfo(path);
if (!file.Exists)
{

FileStream fs;
fs = File.Create(path);
fs.Close();
}
using (FileStream fs = new FileStream(path, FileMode.Append, FileAccess.Write))
{

using (StreamWriter sw = new StreamWriter(fs))
{

sw.WriteLine(DateTime.Now.ToString() + ” ” + msg);
}
}
}
private void PubMessage(string topic, string msg)
{

if (mqttServer != null)
{

lock (locker)
{

var message = new MqttApplicationMessageBuilder();
message.WithTopic(topic);
message.WithPayload(msg);
mqttServer.PublishAsync(message.Build());
}
}
}
/// <summary>
///客户端链接日志客户端接入
/// </summary>
/// <param name=”msg”></param>
private void WriteClientLinkLog(string msg)
{

//string path = @”C:\log.txt”;

//该日志文件会存在windows服务程序目录下
string path = AppDomain.CurrentDomain.BaseDirectory + “\\ClientLinklog.txt”;
FileInfo file = new FileInfo(path);
if (!file.Exists)
{

FileStream fs;
fs = File.Create(path);
fs.Close();
}

using (FileStream fs = new FileStream(path, FileMode.Append, FileAccess.Write))
{

using (StreamWriter sw = new StreamWriter(fs))
{

sw.WriteLine(msg);
}
}
}
/// <summary>
/// 通过定时器将客户端链接信息写入日志
/// </summary>
/// <param name=”sender”></param>
/// <param name=”e”></param>
private void GetSubClientSAndSetShow(object sender, ElapsedEventArgs e)
{

// List<SetServiceM> dic = new List<SetServiceM>();
if (mqttServer != null)
{

List<ConnectedMqttClient> subclients = mqttServer.GetConnectedClientsAsync().Result.ToList();
if (subclients.Count > 0)
{

string subclientcount = @”客户端接入的总数为：” + (subclients.Count – 1).ToString() + “\r\n”
+ “——————————————————- \r\n”;
WriteClientLinkLog(subclientcount);
PubMessage(“ClientsCount”, (subclients.Count – 1).ToString());
List<string> clientids = new List<string>();
//连接客户端的个数
// dic.Add(SetServiceM.SetService( “ClientCount”, subclients.Count.ToString()));
// var dicclientlink = new Dictionary<string, string>();

foreach (var item in subclients)
{

if (!subClientIDs.Contains(item.ClientId))
{

subClientIDs.Add(item.ClientId);
string msg = @”连接客户端ID:” + item.ClientId + “\r\n”
+ “连接时间：” + DateTime.Now + “\r\n”
+ “协议版本：” + item.ProtocolVersion + “\r\n”
+ “最后收到的非保持活包：” + item.LastNonKeepAlivePacketReceived + “\r\n”
+ “最后收到的包：” + item.LastPacketReceived + “\r\n”
+ “挂起的应用程序消息：” + item.PendingApplicationMessages + “\r\n”
+ “————————————————” + “\r\n”;
WriteClientLinkLog(msg);
PubMessage(“clientlink”, msg);
// mqttServer.PublishAsync(“clientlink”, msg);
// dicclientlink.Add(item.ClientId, msg);
}
clientids.Add(item.ClientId);
}
if (subClientIDs.Count >= 2000)
{

subClientIDs.Clear();
}
var exceptlist = subClientIDs.Except(clientids).ToList();
// var dicclientoutline = new Dictionary<string, string>();
if (exceptlist.Count > 0)
{

exceptlist.ForEach(u =>
{

string msgoutline = @”客户端下线ID:” + u + “\r\n”
+ “客户端下线时间：” + DateTime.Now.ToString() + “\r\n”
+ “———————————————————— \r\n”
;
WriteClientLinkLog(msgoutline);
subClientIDs.Remove(u);
PubMessage(“clientlink”, msgoutline);
// mqttServer.PublishAsync(“clientlink”, msgoutline);
// dicclientoutline.Add(“OutLineID_” + u, msgoutline);
});
}
连接客户端的id
//dic.Add(SetServiceM.SetService(“clientlink”, JsonConvert.SerializeObject(dicclientlink)));
客户端下线的时间
//dic.Add(SetServiceM.SetService(“clientoutline”, JsonConvert.SerializeObject(dicclientoutline)));
//SocketServer.connection.Send(Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(dic)));
}
else
{

string subclientcount = @”暂无客户端接入！” + “\r\n”
+ “——————————————————– \r\n”;
WriteClientLinkLog(subclientcount);
}
}
}
/// <summary>
/// 客户端下线时间
/// </summary>
/// <param name=”msg”></param>
public void WriteClientOutLineLog(string msg)
{

string path = AppDomain.CurrentDomain.BaseDirectory + “\\ClientOutLineLog.txt”;
FileInfo file = new FileInfo(path);
if (!file.Exists)
{

FileStream fs = File.Create(path);
fs.Close();
}
using (FileStream fs = new FileStream(path, FileMode.Append, FileAccess.Write))
{

using (StreamWriter sw = new StreamWriter(fs))
{

sw.WriteLine(msg);
}
}
}
//windows服务里的服务端
private void CreateServerSocket()
{

if (socketService == null)
{

// IPEndPoint ipep = new IPEndPoint(IPAddress.Parse(“127.0.0.1”), 9001);
socketService = new GodSharp.Sockets.SocketServer(“127.0.0.1”, 9001, ProtocolType.Tcp); //Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
socketService.Start();
socketService.Listen(10);
Thread thread = new Thread(new ThreadStart(new Action(() =>
{

while (true)
{

// socketClient = socketService.Clients[0];
// string data = “sql|” ; //在这里封装数据，通常是自己定义一种数据结构，如struct data{sql;result}
// client.Send(Encoding.Default.GetBytes(msg));
}
})));
}
else
{

CreateServerSocket();
}
}

#endregion
}

}

服务端桌面显示程序：

using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Diagnostics;
using System.Drawing;
using System.Linq;
using System.Net;
using System.Net.NetworkInformation;
using System.Net.Sockets;
using System.ServiceProcess;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using System.Windows.Forms;

namespace MQTTNetFrm
{

public partial class Form1 : Form
{

private ServiceController ServiceController = null;
private MqttClientOptions options = null;
private MqttClient mqttClient = null;

public Form1()
{

InitializeComponent();

new Thread(new ThreadStart(GetServiceStatus)).Start();

Task.Run(LinkClientService).Wait();
}
/// <summary>
/// 获取当前ip地址
/// </summary>
/// <returns></returns>
private string GetLocalIP()
{

string ip = null;
var iplist = Dns.GetHostAddresses(Dns.GetHostName()).DefaultIfEmpty().ToList();
iplist.ForEach(u =>
{

if (u.AddressFamily == AddressFamily.InterNetwork)
ip= u.ToString();
});
return ip;
}
private async Task LinkClientService()
{

var m = “Eohi_Frm_” + Guid.NewGuid().ToString();
options = new MqttClientOptions
{

ClientId = m,
CleanSession = true,
ChannelOptions = new MqttClientTcpOptions
{

Server = GetLocalIP(),
Port = 1884,
},
Credentials = new MqttClientCredentials()
{

Username = “user”,
Password = “123456”
}

};
var factory = new MqttFactory();
mqttClient = factory.CreateMqttClient() as MqttClient;
try
{

await mqttClient.ConnectAsync(options);
but_submsg_Click();
this.Invoke(new Action(() => { lab_serverstatus.Text = “连接正常，服务运行中…………”; }));
}
catch (Exception ex)
{

}

}
private async void but_submsg_Click()
{

if (mqttClient != null)
{

await mqttClient.SubscribeAsync(new TopicFilter(“ClientsCount”, MqttQualityOfServiceLevel.AtMostOnce));
await mqttClient.SubscribeAsync(new TopicFilter(“clientlink”, MqttQualityOfServiceLevel.AtMostOnce));
await mqttClient.SubscribeAsync(new TopicFilter(“msglog”, MqttQualityOfServiceLevel.AtMostOnce));
mqttClient.ApplicationMessageReceived += (s, e) =>
{

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

var msg = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
if (msg.Length<=5)
{

lab_clientcount.Text = msg;
}
if (msg.Length>10)
{

if (msg.StartsWith(“连接”) )
rtb_clientlog.AppendText(msg);
rtb_msglog.AppendText(msg);
}

}));
};

}
}
private void GetServiceStatus()
{

ServiceController[] serviceControllers = ServiceController.GetServices();
if (serviceControllers.Length > 0)
{

serviceControllers.ToList().ForEach(u =>
{

if (u.DisplayName == “MqttNetServiceAddUserAndPassword”)
{

if (ServiceController == null)
{

ServiceController = u;
}
if (u.Status == ServiceControllerStatus.Running)
{

lab_serverstatus.Text = “服务运行中…………”;
}
else
{

lab_serverstatus.Text = “服务已停止…………”;
}
}
});
}
}
private void button2_Click(object sender, EventArgs e)
{

if (tabControl1.SelectedTab == tabPage1)
{

rtb_clientlog.Text = “”;
}
else
{

rtb_msglog.Text = “”;
}
}

private void Form1_Load(object sender, EventArgs e)
{

}
}

}

客户端：

namespace MqttClientTest01
{

public partial class Form1 : Form
{

private MqttClient mqttClient = null;
private System.Timers.Timer timer = null;
private int CountLink = 0;
private MqttClientOptions options = null;
public Form1()
{

InitializeComponent();
创建一个定时器，检查5s内有多少客户端接入并将相关信息记录到日志中
//timer = new System.Timers.Timer();
//timer.AutoReset = true;
//timer.Interval = 1000;
//timer.Elapsed += new ElapsedEventHandler(LinkMqttNetService);
}

private void LinkMqttNetService(object sender, ElapsedEventArgs e)
{

if (mqttClient == null)
{

// RunAsync();
CountLink++;
}
if (CountLink >= 5)
{

MessageBox.Show(“连接多次失败，请确认各参数是否正确！”);
CountLink = 0;
timer.Enabled = false;
}
}
private void but_linkserver_Click(object sender, EventArgs k)
{

LinkClientService();
//CountLink = 0;
//timer.Enabled = true;
//timer.Start();
}
/// <summary>
/// 链接客户端
/// </summary>
public async void LinkClientService()
{

var m = “Eohi_” + Guid.NewGuid().ToString();
options = new MqttClientOptions
{

ClientId = m,
CleanSession = true,
ChannelOptions = new MqttClientTcpOptions
{

Server = txtb_serverip.Text.Trim(),
Port = Convert.ToInt32(txtb_serverport.Text.Trim()),
},
Credentials = new MqttClientCredentials()
{

Username = tb_username.Text,
Password = tb_userpwd.Text
}

};
var factory = new MqttFactory();
mqttClient = factory.CreateMqttClient() as MqttClient;
try
{

await mqttClient.ConnectAsync(options);
this.Invoke(new Action(() =>
{

lab_linkstatus.Text = “连接成功！”;
lab_linktimer.Text = DateTime.Now.ToString();
}));
mqttClient.Disconnected += async (s, e) =>
{

if (e.ClientWasConnected==false)
{

try
{

await mqttClient.ConnectAsync(options);
this.Invoke(new Action(() =>
{

lab_linkstatus.Text = “连接成功！”;
lab_linktimer.Text = DateTime.Now.ToString();
}));
}
catch (Exception ex)
{

lab_linkstatus.Text = “连接失败！”+ex.Message;
lab_linktimer.Text = DateTime.Now.ToString();
}

}
};
}
catch (Exception ex)
{

lab_linkstatus.Text = “连接失败！请检查ip/端口” ;
lab_linktimer.Text = DateTime.Now.ToString();
}

}

private void tb_username_TextChanged(object sender, EventArgs e)
{

}

private void but_clientsend_Click(object sender, EventArgs e)
{

if (mqttClient != null)
{

var message = new MqttApplicationMessageBuilder();
message.WithTopic(txtb_msgtopic.Text.Trim());
message.WithPayload(rtb_pubmsg.Text.Trim());
message.WithExactlyOnceQoS();
message.WithRetainFlag();
mqttClient.PublishAsync(message.Build());
}
}
private async void but_submsg_Click(object sender, EventArgs k)
{

if (mqttClient != null)
{

await mqttClient.SubscribeAsync(new TopicFilter(txtb_subtopic.Text.Trim(), MqttQualityOfServiceLevel.AtMostOnce));
mqttClient.ApplicationMessageReceived += (s, e) =>
{

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

rtb_submsgclient.AppendText(“ClientID=” + e.ClientId + “\n”);
rtb_submsgclient.AppendText($”+ Topic = {e.ApplicationMessage.Topic}” + “\n”);
rtb_submsgclient.AppendText($”+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload) + “\n”}”);
rtb_submsgclient.AppendText($”+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}” + “\n”);
rtb_submsgclient.AppendText($”+ Retain = {e.ApplicationMessage.Retain}” + “\n”);