Understanding and applying the MQTT protocol is a crucial skill in the field of the Internet of Things. To ensure stable connections between devices and servers, we need to deeply understand and effectively apply the automatic reconnection feature of MQTT clients. Next, let's delve into the MQTT protocol and reconnection mechanism, just like exploring a mysterious adventure island.
- Navigational keep-alive timeIn MQTT, we refer to it as Keep Alive. It is a timer that periodically checks whether our "ship" maintains a connection with the "port". We need to set a suitable Keep Alive based on the actual network environment and application requirements.
- Reconnection strategy and backoffWhen our "ship" loses connection with the "port", we should not immediately attempt to reconnect. Instead, we should set a reasonable waiting time to avoid excessive resource consumption. This is like when our ship is lost at sea, we need to temporarily stop, observe the wind direction, measure the current, and then devise a new route. We can use an exponential backoff algorithm or a stepwise delay strategy to achieve this function.
- Connection status managementOur "ship" requires a logbook to record important information such as the connection status with the "port", the reason for disconnection, and subscribed information. When the connection is disconnected, our "ship" should refer to the logbook, analyze the reason for disconnection, and then attempt to reconnect with the "port".
- exception handlingDuring the voyage, our "ship" may encounter various issues, such as "port" unavailability, authentication failure, network anomalies, etc. Our "ship" needs to have an emergency plan to deal with these issues. For example, when the "port" is unavailable, our "ship" may need to seek alternative "ports"; when authentication fails, our "ship" may need to check whether its own authentication information is correct; when there is a network anomaly, our "ship" may need to suspend its voyage and wait for the network to return to normal.
- Maximum attempt limitFor some low-power devices, we may need to consider limiting the number of reconnection attempts to avoid excessively draining the device's battery. Just like a "ship" lost at sea, when it has tried many times but failed to find a "port", it may need to temporarily stop and wait for better sailing conditions.
After designing this automatic navigation system (the automatic reconnection logic of the MQTT client), our "ship" can navigate the ocean of the Internet of Things more effectively. No matter what challenges it faces, it can always maintain a stable connection with the "port", thereby ensuring the smooth operation of our application.
/*******************************************************************************
* 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;
}
Leave a Reply