大家好，又见面了，我是你们的朋友全栈君。
1 ///开源库地址：https://github.com/chkr1011/MQTTnet
2 ///对应文档：https://github.com/chkr1011/MQTTnet/wiki/Client
3
4 usingMQTTnet;5 usingMQTTnet.Client;6 usingMQTTnet.Client.Options;7 usingSystem;8 usingSystem.Text;9 usingSystem.Threading;10 usingSystem.Threading.Tasks;11 usingSystem.Windows.Forms;12
13 namespaceMqttServerTest14 {15 public partial classmqtt测试工具 : Form16 {17 private IMqttClient mqttClient = null;18 private bool isReconnect = true;19
20 publicmqtt测试工具()21 {22 InitializeComponent();23 }24
25 private void Form1_Load(objectsender, EventArgs e)26 {27
28 }29
30 private async void BtnPublish_Click(objectsender, EventArgs e)31 {32 awaitPublish();33 }34
35 private async void BtnSubscribe_ClickAsync(objectsender, EventArgs e)36 {37 awaitSubscribe();38 }39
40 private asyncTask Publish()41 {42 string topic =txtPubTopic.Text.Trim();43
44 if (string.IsNullOrEmpty(topic))45 {46 MessageBox.Show(“发布主题不能为空！”);47 return;48 }49
50 string inputString =txtSendMessage.Text.Trim();51 try
52 {53
54 var message = newMqttApplicationMessageBuilder()55 .WithTopic(topic)56 .WithPayload(inputString)57 .WithExactlyOnceQoS()58 .WithRetainFlag()59 .Build();60
61 awaitmqttClient.PublishAsync(message);62 }63 catch(Exception ex)64 {65
66 Invoke((new Action(() =>
67 {68 txtReceiveMessage.AppendText($”发布主题失败！” Environment.NewLine ex.Message Environment.NewLine);69 })));70 }71
72
73
74
75 }76
77 private asyncTask Subscribe()78 {79 string topic =txtSubTopic.Text.Trim();80
81 if (string.IsNullOrEmpty(topic))82 {83 MessageBox.Show(“订阅主题不能为空！”);84 return;85 }86
87 if (!mqttClient.IsConnected)88 {89 MessageBox.Show(“MQTT客户端尚未连接！”);90 return;91 }92
93 //Subscribe to a topic
94 await mqttClient.SubscribeAsync(newTopicFilterBuilder()95 .WithTopic(topic)96 .WithAtMostOnceQoS()97 .Build()98 );99 Invoke((new Action(() =>
100 {101 txtReceiveMessage.AppendText($”已订阅[{topic}]主题{Environment.NewLine}”);102 })));103
104 }105
106 private asyncTask ConnectMqttServerAsync()107 {108 //Create a new MQTT client.
109
110 if (mqttClient == null)111 {112 try
113 {114 var factory = newMqttFactory();115 mqttClient =factory.CreateMqttClient();116
117 var options = newMqttClientOptionsBuilder()118 .WithTcpServer(txtIp.Text, Convert.ToInt32(txtPort.Text)).WithCredentials(txtUsername.Text, txtPsw.Text).WithClientId(txtClientId.Text) //Port is optional
119 .Build();120
121
122 awaitmqttClient.ConnectAsync(options, CancellationToken.None);123 Invoke((new Action(() =>
124 {125 txtReceiveMessage.AppendText($”连接到MQTT服务器成功！” txtIp.Text);126 })));127 mqttClient.UseApplicationMessageReceivedHandler(e =>
128 {129
130 Invoke((new Action(() =>
131 {132 txtReceiveMessage.AppendText($”收到订阅消息！” Encoding.UTF8.GetString(e.ApplicationMessage.Payload));133 })));134
135 });136 }137 catch(Exception ex)138 {139
140 Invoke((new Action(() =>
141 {142 txtReceiveMessage.AppendText($”连接到MQTT服务器失败！” Environment.NewLine ex.Message Environment.NewLine);143 })));144 }145 }146 }147
148 private void MqttClient_Connected(objectsender, EventArgs e)149 {150 Invoke((new Action(() =>
151 {152 txtReceiveMessage.Clear();153 txtReceiveMessage.AppendText(“已连接到MQTT服务器！” Environment.NewLine);154 })));155 }156
157 private void MqttClient_Disconnected(objectsender, EventArgs e)158 {159 Invoke((new Action(() =>
160 {161 txtReceiveMessage.Clear();162 DateTime curTime = newDateTime();163 curTime =DateTime.UtcNow;164 txtReceiveMessage.AppendText($”>> [{curTime.ToLongTimeString()}]”);165 txtReceiveMessage.AppendText(“已断开MQTT连接！” Environment.NewLine);166 })));167
168 //Reconnecting
169 if(isReconnect)170 {171 Invoke((new Action(() =>
172 {173 txtReceiveMessage.AppendText(“正在尝试重新连接” Environment.NewLine);174 })));175
176 var options = newMqttClientOptionsBuilder()177 .WithClientId(txtClientId.Text)178 .WithTcpServer(txtIp.Text, Convert.ToInt32(txtPort.Text))179 .WithCredentials(txtUsername.Text, txtPsw.Text)180 //.WithTls()
181 .WithCleanSession()182 .Build();183 Invoke((new Action(async () =>
184 {185 await Task.Delay(TimeSpan.FromSeconds(5));186 try
187 {188 awaitmqttClient.ConnectAsync(options);189 }190 catch
191 {192 txtReceiveMessage.AppendText(“### RECONNECTING FAILED ###” Environment.NewLine);193 }194 })));195 }196 else
197 {198 Invoke((new Action(() =>
199 {200 txtReceiveMessage.AppendText(“已下线！” Environment.NewLine);201 })));202 }203 }204
205 private void MqttClient_ApplicationMessageReceived(objectsender, MqttApplicationMessageReceivedEventArgs e)206 {207 Invoke((new Action(() =>
208 {209 txtReceiveMessage.AppendText($”>> {“### RECEIVED APPLICATION MESSAGE ###”}{Environment.NewLine}”);210 })));211 Invoke((new Action(() =>
212 {213 txtReceiveMessage.AppendText($”>> Topic = {e.ApplicationMessage.Topic}{Environment.NewLine}”);214 })));215 Invoke((new Action(() =>
216 {217 txtReceiveMessage.AppendText($”>> Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}{Environment.NewLine}”);218 })));219 Invoke((new Action(() =>
220 {221 txtReceiveMessage.AppendText($”>> QoS = {e.ApplicationMessage.QualityOfServiceLevel}{Environment.NewLine}”);222 })));223 Invoke((new Action(() =>
224 {225 txtReceiveMessage.AppendText($”>> Retain = {e.ApplicationMessage.Retain}{Environment.NewLine}”);226 })));227 }228
229 private void btnLogIn_Click(objectsender, EventArgs e)230 {231 isReconnect = true;232 Task.Run(async () => { awaitConnectMqttServerAsync(); });233 }234
235 private void btnLogout_Click(objectsender, EventArgs e)236 {237 isReconnect = false;238 Task.Run(async () => { awaitmqttClient.DisconnectAsync(); });239 }240
241 }242 }
发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/152856.html原文链接：https://javaforall.cn
