我是不会的，全是从网上搜的，最终整理拼合的可以使用了，使用C#和VS2019，MQTT使用3.1.0版本，需要注意的是不同的版本代码是不一样的，对于咱这种不会的，当然是以能用为主了，你要是安装的最新的4.0版本，那还是换换吧
首先需要在nuGet中引用下面的三个，MQTTnet 3.1.0、MQTTnet.Extensions.ManagedClient 3.1.0、Newtonsoft.json 10.0.1，如果不引用的话，代码中会有错误，自己还解决不了，白白浪费你时间
代码是可以完全跑起来的，我在本机上已经运行过了，.net 版本是4.5.2
一、服务端
服务端主要是设置你要监听哪个端口，是否要验证客户端的用户名和密码，然后开启服务监听就行了，代码里设置的用户名是test，密码是1234，可以根据实际换成数据库中的用户名密码来验证，对于接收到的消息，现在是直接显示，以后可以保存到日志文件中，也可以保存到数据库中。
using MQTTnet.Server;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;
using MQTTnet.Client.Receiving;
namespace MQTT
{
public partial class Form1 : Form
{
public Form1()
{
InitializeComponent();
}
int Port = 2023;
IMqttServer server = new MqttFactory().CreateMqttServer();
private void Form1_Load(object sender, EventArgs e)
{
}
private void btn_start_Click(object sender, EventArgs e)
{
StartMQTTAsync();
}
//启动服务端
public async Task StartMQTTAsync()
{
MqttServerOptionsBuilder serverOptions = new MqttServerOptionsBuilder();
serverOptions.WithConnectionValidator(client =>
{
string Account = client.Username;
string PassWord = client.Password;
string clientid = client.ClientId;
if (Account == "test" && PassWord == "1234")
{
client.ReasonCode = MqttConnectReasonCode.Success;
Console.WriteLine("校验成功");
}
else
{
client.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
Console.WriteLine("校验失败");
}
});
serverOptions.WithDefaultEndpointPort(Port);
//服务启动
server.StartedHandler = new MqttServerStartedHandlerDelegate((Action<EventArgs>)StartedHandler);
//服务停止
server.StoppedHandler = new MqttServerStoppedHandlerDelegate((Action<EventArgs>)StoppedHandler);
//客户端连接事件
server.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate((Action<MqttServerClientConnectedEventArgs>)ClientConnectedHandler);
//客户端断开连接事件
server.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate((Action<MqttServerClientDisconnectedEventArgs>)ClientDisconnectedHandler);
//消息监听
server.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate((Action<MqttApplicationMessageReceivedEventArgs>)MessageReceivedHandler);
//客户端订阅主题事件
server.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate((Action<MqttServerClientSubscribedTopicEventArgs>)ClientSubscribedTopicHandler);
//客户端取消订阅主题事件
server.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate((Action<MqttServerClientUnsubscribedTopicEventArgs>)ClientUnsubscribedTopicHandler);
await server.StartAsync(serverOptions.Build());
}
public void StartedHandler(EventArgs obj)
{
L1.Items.Add("MQTT程序已启动，监听端口："+Port.ToString());
}
/// <summary>
/// MQTT服务器停止事件
/// </summary>
/// <param name="obj"></param>
private void StoppedHandler(EventArgs obj)
{
L1.Items.Add("MQTT程序已经关闭！");
}
/// <summary>
/// 客户端连接到服务器事件
/// </summary>
/// <param name="obj"></param>
private void ClientConnectedHandler(MqttServerClientConnectedEventArgs obj)
{
L1.Items.Add($"{obj.ClientId}此客户端已经连接到服务器");
}
/// <summary>
/// 客户端断开连接事件
/// </summary>
/// <param name="obj"></param>
private void ClientDisconnectedHandler(MqttServerClientDisconnectedEventArgs obj)
{
L1.Items.Add($"断开连接的客户端:{obj.ClientId}");
L1.Items.Add($"断开连接类型:{obj.DisconnectType.ToString()}");
}
/// <summary>
/// 收到各个客户端发送的消息
/// </summary>
/// <param name="obj"></param>
private void MessageReceivedHandler(MqttApplicationMessageReceivedEventArgs obj)
{
L1.Items.Add("===================================================");
L1.Items.Add("收到消息:");
L1.Items.Add("消息时间：" + DateTime.Now.ToString()); ;
L1.Items.Add($"客户端:{obj.ClientId}");
L1.Items.Add($"主题:{obj.ApplicationMessage.Topic}");
L1.Items.Add($"消息:{Encoding.UTF8.GetString(obj.ApplicationMessage.Payload)}");
L1.Items.Add("+++++++++++++++++++++++++++++++++++++++++++++++++++");
L1.Items.Add("");
}
/// <summary>
/// 客户端订阅的主题
/// </summary>
/// <param name="obj"></param>
private void ClientSubscribedTopicHandler(MqttServerClientSubscribedTopicEventArgs obj)
{
L1.Items.Add($"客户端:{obj.ClientId}");
L1.Items.Add($"订阅主题:{obj.TopicFilter.Topic}");
}
/// <summary>
/// 客户端取消订阅主题
/// </summary>
/// <param name="obj"></param>
private void ClientUnsubscribedTopicHandler(MqttServerClientUnsubscribedTopicEventArgs obj)
{
L1.Items.Add($"客户端:{obj.ClientId}");
L1.Items.Add($"取消订阅主题:{obj.TopicFilter}");
}
/// <summary>
/// 关闭服务
/// </summary>
/// <returns></returns>
public async Task StopAsync()
{
if (server != null)
{
if (server.IsStarted)
{
await server.StopAsync();
server.Dispose();
}
}
}
private void btn_stop_Click(object sender, EventArgs e)
{
StopAsync();
}
}
}
二、客户端
客户端就是设置服务端的地址和端口，用户名和密码，然后订阅主题，然后发送消息，咱们不会，就用各种框框来直观的操作了，可根据需要设计成简洁界面的，更方便使用
不知道是客户端ID不能重复还是不能在一台电脑上开两个客户端，反正开两个客户端后会反复的连接和断开，留给你们试吧
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Client.Receiving;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Formatter;
using Newtonsoft.Json;
namespace MQTT
{
public partial class Form2 : Form
{
public Form2()
{
InitializeComponent();
}
private MqttClientOptions options;
private IManagedMqttClient mqttClient;
private void Form2_Load(object sender, EventArgs e)
{
CheckForIllegalCrossThreadCalls = false;
}
private async void OnSubscriberConnected(MqttClientConnectedEventArgs x)
{
L_info.Items.Add("已连接到MQTT服务器！");
}
private async void OnSubscriberDisconnected(MqttClientDisconnectedEventArgs x)
{
L_info.Items.Add("已断开MQTT服务器连接！");
}
private void OnSubscriberMessageReceived(MqttApplicationMessageReceivedEventArgs x)
{
var payloadString = x.ApplicationMessage.ConvertPayloadToString();
payloadString = ConvertJsonString(payloadString);
var item = $"{Environment.NewLine}Topic: {x.ApplicationMessage.Topic}{Environment.NewLine}Payload: {payloadString} {Environment.NewLine}QoS: {x.ApplicationMessage.QualityOfServiceLevel}";
this.BeginInvoke((MethodInvoker)delegate
{
AppendReceiveMsg(item);
});
}
private async Task SubscriberStart()
{
var tcpServer = txt_ip.Text.Trim().ToString();
var tcpPort = int.Parse(txt_port.Text.Trim());
var mqttUser = txt_yhm.Text.Trim();
var mqttPassword = txt_pwd.Text.Trim();
var mqttFactory = new MqttFactory();
this.options = new MqttClientOptions
{
ClientId = "Client-1",
ProtocolVersion = MqttProtocolVersion.V311,
ChannelOptions = new MqttClientTcpOptions
{
Server = tcpServer,
Port = tcpPort
}
};
if (options.ChannelOptions == null)
{
throw new InvalidOperationException();
}
if (!string.IsNullOrEmpty(mqttUser))
{
options.Credentials = new MqttClientCredentials
{
Username = mqttUser,
Password = Encoding.UTF8.GetBytes(mqttPassword)
};
}
options.CleanSession = true;
options.KeepAlivePeriod = TimeSpan.FromSeconds(5);
this.mqttClient = mqttFactory.CreateManagedMqttClient();
this.mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate((Action<MqttClientConnectedEventArgs>)OnSubscriberConnected);
this.mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate((Action<MqttClientDisconnectedEventArgs>)OnSubscriberDisconnected);
this.mqttClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate((Action<MqttApplicationMessageReceivedEventArgs>)OnSubscriberMessageReceived);
await this.mqttClient.StartAsync(
new ManagedMqttClientOptions
{
ClientOptions = options
});
}
/// <summary>
/// 发送MQTT消息
/// </summary>
/// <param name="dyh">订阅号</param>
/// <param name="msg">具体发送的消息</param>
private async void sengMsg(string dyh,string msg)
{
var publish_topic = dyh;
var publish_msg = msg;
var message = new MqttApplicationMessageBuilder()
.WithTopic(publish_topic)
.WithPayload(publish_msg)
.WithExactlyOnceQoS()
.Build();
if (this.mqttClient != null)
{
await this.mqttClient.PublishAsync(message);
}
}
private void AppendReceiveMsg(string msg)
{
Invoke((new Action(() =>
{
L_info.Items.Add(DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss: ") + msg + Environment.NewLine + Environment.NewLine);
})));
}
private void AppendSendMsg(string msg)
{
Invoke((new Action(() =>
{
L_info.Items.Add(DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss: ") + msg + Environment.NewLine);
})));
}
private void AppendLogMsg(string msg)
{
Invoke((new Action(() =>
{
L_info.Items.Add(DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss: ") + msg + Environment.NewLine);
})));
}
private string ConvertJsonString(string str)
{
try
{
//格式化json字符串
JsonSerializer serializer = new JsonSerializer();
TextReader tr = new StringReader(str);
JsonTextReader jtr = new JsonTextReader(tr);
object obj = serializer.Deserialize(jtr);
if (obj != null)
{
StringWriter textWriter = new StringWriter();
JsonTextWriter jsonWriter = new JsonTextWriter(textWriter)
{
Formatting = Formatting.Indented,
Indentation = 4,
IndentChar = ' '
};
serializer.Serialize(jsonWriter, obj);
return textWriter.ToString();
}
return str;
}
catch (Exception ex)
{
return str;
}
}
private async void btn_open_Click(object sender, EventArgs e)
{
//打开MQTT连接
if (this.mqttClient == null)
{
await SubscriberStart();
}
}
private async void btn_close_Click(object sender, EventArgs e)
{
if (this.mqttClient == null)
{
return;
}
await this.mqttClient.StopAsync();
this.mqttClient = null;
}
private async void btn_dingyue_Click(object sender, EventArgs e)
{
var topicFilter = new MqttTopicFilter { Topic = this.txt_dyh.Text.Trim() };
await this.mqttClient.SubscribeAsync(topicFilter);
L_info.Items.Add("已订阅消息！");
}
private void btn_quxiao_Click(object sender, EventArgs e)
{
}
private void btn_send_Click(object sender, EventArgs e)
{
string dyh = txt_dyh.Text.Trim();
string msg = txt_msg.Text.Trim();
sengMsg(dyh, msg);
}
private void groupBox4_Enter(object sender, EventArgs e)
{
}
}
}
最后，程序和你肯定有一个能跑起来的，祝你好运~
标签：Items,void,Add,private,发送,MQTT,new,using,接收 From： https://www.cnblogs.com/wjbych/p/17474650.html