Das Verständnis und die Anwendung des MQTT-Protokolls ist eine wichtige Fähigkeit im Internet der Dinge. Um eine stabile Verbindung zwischen Geräten und Servern zu gewährleisten, müssen wir die automatischen Wiederverbindungsfunktionen des MQTT-Clients besser verstehen und effektiv anwenden. Im Folgenden lassen Sie uns das MQTT-Protokoll und die Wiederverbindungsmechanismen genauso vertiefen, wie Sie eine mysteriöse Abenteuerinsel erkunden. MQTT ist ein Publish / Subscribe-Modell - Protokoll, das auf dem TCP-Protokoll basiert. Es ist wie ein durch Wind und Regen getaufenes Schiff, das durch das Internet der Dinge, Sensornetzwerke und andere unstabile Netzwerkumgebungen mit geringer Bandbreite segelt. Aber wie die Wellen auf dem Meer, ist auch die Netzwerkumgebung voller Herausforderungen: Netzwerk-Ausfälle, Signalschwäche, Datenpaketverlust und so weiter, die die Verbindung zwischen MQTT-Clients und Servern unterbrechen können. In den Ozeanen des Internet der Dinge sind häufige Wellen, die die Wiederverbindung auslösen, schlechte Netzwerkumgebungen oder Netzwerk-Ausfälle, Server-Upgrades, Neustart von Geräten oder Clients und andere Netzwerkfaktoren. Wie können wir in diesem Fall sicherstellen, dass unser "Schiff" (MQTT-Client) immer eine stabile Verbindung mit dem "Hafen" (Server) hat? Die Antwort war, dass wir unser "Schiff" mit einem automatischen Navigationssystem ausstatten mussten - der automatischen Wiedereinbindungslogik des MQTT-Clients. Das Entwerfen eines ausgezeichneten MQTT-Clients ist wie der Bau eines robusten Seeschiffes. Wenn das Design nicht vernünftig ist, kann unser "Schiff" die Navigation verlieren, die Stille nicht mehr von dem "Hafen" Nachrichten empfangen, und kann sogar unbewusst unseren "Hafen" angreifen, weil sie häufig versuchen, die Verbindung wieder herzustellen, das ist wie ein Kopfloses Rennen auf dem Meer, das nicht nur seine eigene Energie verbraucht, sondern auch unnötigen Druck auf den "Hafen" bringt. Wenn unsere Logik für die Wiederverbindung jedoch vernünftig ausgelegt ist, kann unser "Schiff" automatisch steuern, den "Hafen" wiederfinden und die Verbindung zu ihm halten, wann immer die Verbindung verloren geht. Bei der Gestaltung der MQTT-Client - Wiederverbindungslogik müssen wir einige wichtige Faktoren berücksichtigen:
- Navigation-Laufzeit: In MQTT bezeichnen wir dies als Keep Alive. Dies ist ein Timer, der regelmäßig überprüft, ob unser "Schiff" mit dem "Hafen" verbunden ist. Wir müssen ein geeignetes Keep Alive auf der Grundlage der tatsächlichen Netzwerkumgebung und Anwendungsanforderungen einrichten.
- Wiederverbindungsstrategie und Rückzug: Wenn unser "Schiff" die Verbindung mit dem "Hafen" verliert, sollten wir nicht sofort versuchen, eine Wiederverbindung zu erstellen, sondern eine angemessene Wartezeit festlegen, um übermäßigen Verbrauch von Ressourcen zu vermeiden. Es ist wie wenn unser Schiff auf See verloren geht, müssen wir eine Pause machen, um die Richtung des Windes zu beobachten, die Strömung zu messen, bevor wir eine neue Route erstellen. Wir können einen exponentiellen Rücktritt-Algorithmus oder eine Stufen-Verzögerungsstrategie verwenden, um diese Funktion zu implementieren.
- Connection Status Management: Unser Schiff benötigt ein Navigationsprotokoll, um wichtige Informationen wie den Verbindungsstatus mit dem Hafen, die Gründe für die Verbindungsabschaltung, bereits abonnierte Informationen und so weiter aufzuzeichnen. Wenn die Verbindung unterbrochen ist, sollte unser "Schiff" das Navigationsprotokoll überprüfen, die Ursachen der Verbindung unterbrochen zu analysieren und dann versuchen, den "Hafen" wieder zu verbinden.
- Außergewöhnliche Behandlung: Während der Fahrt kann unser "Schiff" auf verschiedene Probleme stoßen, wie zum Beispiel nicht verfügbare "Hafen", fehlgeschlagenes Zertifizieren, Netzwerk-Ausnahmen usw. Unser „Schiff" braucht einen Notfallplan, um auf diese Probleme zu reagieren. Zum Beispiel, wenn der "Hafen" nicht verfügbar ist, muss unser "Schiff" möglicherweise einen anderen "Hafen" suchen; Wenn die Authentifizierung fehlschlägt, muss unser "Schiff" möglicherweise überprüfen, ob seine eigenen Authentifizierungsinformationen korrekt sind; Wenn das Netzwerk abnormal ist, muss unser "Schiff" möglicherweise die Fahrt ausbrechen und warten, bis das Netzwerk wieder normal ist.
- Maximale Anzahl der Versuche begrenzt: Bei einigen Geräten mit niedrigem Stromverbrauch müssen wir möglicherweise in Betracht ziehen, die Anzahl der Wiederverbindungsversuche zu begrenzen, um einen übermäßigen Stromverbrauch des Geräts zu vermeiden. Genau wie ein "Schiff", das auf dem Meer wandert, wenn es viele Versuche, den "Hafen" nicht zu finden, muss es möglicherweise vorübergehend stoppen, um auf bessere Segelbedingungen zu warten.
Mit der Entwicklung dieses automatischen Navigationssystems (automatische Wiederverbindungslogik des MQTT-Clients) ist unser „Schiff" in der Lage, im Ozean des IoT besser zu navigieren, unabhängig von den Herausforderungen, immer eine stabile Verbindung mit dem „Hafen" zu halten, um sicherzustellen, dass unsere Anwendung reibungslos läuft. Schauen wir uns einen praktischen Fall an. Wir nehmen als Beispiel die Paho MQTT C-Bibliothek, die uns eine umfangreiche Reihe von Navigationswerkzeugen bietet - Callback-Funktionen, die es uns ermöglichen, die Funktionsweise des automatischen Navigationssystems anhand der tatsächlichen Situationen zu konfigurieren. Paho bietet globale Callbacks, API Callbacks und asynchrone Methoden Callbacks, sodass wir in jeder Situation mit dem "Hafen" verbunden bleiben können. Das ist die Geschichte, wie wir im Ozean des Internets der Dinge segeln. Ich hoffe, dass diese Geschichte Ihnen hilft, das MQTT-Protokoll und den Wiederverbindungsmechanismus besser zu verstehen, und ich hoffe, dass Ihr "Schiff" reibungslos in den Ozeanen des Internets der Dinge segelt.
/*******************************************************************************
* Copyright (c) 2012, 2022 IBM Corp., Ian Craggs
*
* 保留所有权利。此程序和随附的资料
* 根据Eclipse公共许可证v2.0
* 和Eclipse发行许可证v1.0的条款提供。
*
* Eclipse公共许可证可在以下网址查阅
* https://www.eclipse.org/legal/epl-2.0/
* Eclipse发行许可证可在以下网址查阅
* http://www.eclipse.org/org/documents/edl-v10.php。
*
*******************************************************************************/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTAsync.h"
#if !defined(_WIN32)
#include <unistd.h>
#else
#include <windows.h>
#endif
#if defined(_WRS_KERNEL)
#include <OsWrapper.h>
#endif
// 定义需要使用的MQTT连接参数，如broker地址和客户端ID等
#define ADDRESS "tcp://broker.emqx.io:1883"
#define CLIENTID "PahoClientSub"
#define TOPIC "nanomq/test"
#define PAYLOAD "Hello World!"
#define QOS 1
#define TIMEOUT 10000L
// 定义在主线程中的逻辑Flag
int disc_finished = 0;
int subscribed = 0;
int finished = 0;
//首先声明 API 回调函数
void onConnect(void* context, MQTTAsync_successData* response);
void onConnectFailure(void* context, MQTTAsync_failureData* response);
void onSubscribe(void* context, MQTTAsync_successData* response);
void onSubscribeFailure(void* context, MQTTAsync_failureData* response);
// 下面2个是 Async 使用的回调函数
// 异步连接成功的回调函数，在连接成功的时候进行Subscribe操作。
void conn_established(void *context, char *cause)
{
printf("客户端已重新连接!n");
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
int rc;
printf("连接成功n");
printf("订阅主题 %sn使用客户端 %s 并用QoS%dnn"
"按Q<Enter>退出nn", TOPIC, CLIENTID, QOS);
opts.onSuccess = onSubscribe;
opts.onFailure = onSubscribeFailure;
opts.context = client;
if ((rc = MQTTAsync_subscribe(client, TOPIC, QOS, &opts)) != MQTTASYNC_SUCCESS)
{
printf("开始订阅失败，返回码 %dn", rc);
finished = 1;
}
}
// 异步连接收到 Disconnect消息时的回调，由于大部分断开的情况下不会收到 Disconnect消息，所以此方法很少被触发
void disconnect_lost(void* context, MQTTProperties* properties,
enum MQTTReasonCodes reasonCode)
{
printf("客户端已断开连接!n");
}
// 下面是客户端全局回调函数，分别是连接断开和消息到达
void conn_lost(void *context, char *cause)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
int rc;
printf("n连接已断开n");
if (cause)
printf(" 原因: %sn", cause);
printf("正在重连n");
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
conn_opts.maxRetryInterval = 16;
conn_opts.minRetryInterval = 2;
conn_opts.automaticReconnect = 1;
//conn_opts.onSuccess = onConnect;
conn_opts.onFailure = onConnectFailure;
MQTTAsync_setConnected(client, client, conn_established);
if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
{
printf("开始连接失败，返回码 %dn", rc);
finished = 1;
}
}
// 收到消息时的全局回调函数，此处简单的打印消息
int msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
printf("消息已到达n");
printf(" 主题: %sn", topicName);
printf(" 消```C
息: ");
/* 打印消息内容 */
char* payloadptr = message->payload;
for(int i = 0; i < message->payloadlen; i++)
{
putchar(*payloadptr++);
}
putchar('n');
/* 释放消息内存 */
MQTTAsync_freeMessage(&message);
MQTTAsync_free(topicName);
return 1;
}
/* 异步断开连接的回调函数 */
void onDisconnect(void* context, MQTTAsync_successData* response)
{
printf("成功断开连接n");
disc_finished = 1;
}
/* 异步连接成功的回调函数，在连接成功的时候进行订阅操作。 */
void onConnect(void* context, MQTTAsync_successData* response)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
int rc;
printf("成功连接n");
printf("订阅主题 %sn使用客户端 %s 并用QoS%dnn"
"按Q<Enter>退出nn", TOPIC, CLIENTID, QOS);
/* 开始订阅 */
opts.onSuccess = onSubscribe;
opts.onFailure = onSubscribeFailure;
opts.context = client;
if ((rc = MQTTAsync_subscribe(client, TOPIC, QOS, &opts)) != MQTTASYNC_SUCCESS)
{
printf("开始订阅失败，返回码 %dn", rc);
finished = 1;
}
}
/* 异步连接失败的回调函数 */
void onConnectFailure(void* context, MQTTAsync_failureData* response)
{
printf("连接失败n");
if (response && response->message)
{
printf("失败信息: %sn", response->message);
}
finished = 1;
}
/* 异步订阅成功的回调函数 */
void onSubscribe(void* context, MQTTAsync_successData* response)
{
printf("成功订阅n");
subscribed = 1;
}
/* 异步订阅失败的回调函数 */
void onSubscribeFailure(void* context, MQTTAsync_failureData* response)
{
printf("订阅失败n");
if (response && response->message)
{
printf("失败信息: %sn", response->message);
}
finished = 1;
}
/* 异步取消订阅的回调函数 */
void onUnsubscribe(void* context, MQTTAsync_successData* response)
{
printf("成功取消订阅n");
finished = 1;
}
int main(int argc, char* argv[])
{
MQTTAsync client;
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
int rc;
MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
MQTTAsync_token token;
/* 创建MQTT客户端 */
MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
/* 设置全局回调函数 */
MQTTAsync_setCallbacks(client, client, conn_lost, msgarrvd, NULL);
/* 设置连接选项 */
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
conn_opts.automaticReconnect = 1;
//conn_opts.onSuccess = onConnect;
conn_opts.onFailure = onConnectFailure;
MQTTAsync_setConnected(client, client, conn_established);
/* 开始连接 */
if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
{
printf("开始连接失败，返回码 %dn", rc);
return EXIT_FAILURE;
}
while (!finished)
{
#if defined(_WIN32)
Sleep(1000);
#else
sleep(1);
#endif
}
if (subscribed)
{
if ((rc = MQTTAsync_unsubscribe(client, TOPIC, NULL)) != MQTTASYNC_SUCCESS)
{
printf("取消订阅失败，返回码 %dn", rc);
return EXIT_FAILURE;
}
}
/* 断开连接 */
MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
disc_opts.onSuccess = onDisconnect;
if ((rc = MQTTAsync_disconnect(client, &disc_opts)) != MQTTASYNC_SUCCESS)
{
printf("开始断开连接失败，返回码 %dn", rc);
return EXIT_FAILURE;
}
while (!disc_finished)
{
#if defined(_WIN32)
Sleep(1000);
#else
sleep(1);
#endif
}
/* 销毁客户端 */
MQTTAsync_destroy(&client);
return EXIT_SUCCESS;
}
Antwort veröffentlichen