MQTT 客户端自动重连实践构建可靠 IoT 设备连接 - 谷动谷力

// 是 Async 使用的回调方法
// 连接成功的异步回调函数，在连接成功的地方进行Subscribe操作。
void conn_established(void *context, char *cause)
{
printf("client reconnected!\n");
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
int rc;
printf("Successful connection\n");
printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
"Press Q<Enter> to quit\n\n", TOPIC, CLIENTID, QOS);
opts.onSuccess = onSubscribe;
opts.onFailure = onSubscribeFailure;
opts.context = client;
if ((rc = MQTTAsync_subscribe(client, TOPIC, QOS, &opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start subscribe, return code %d\n", rc);
finished = 1;
}
}
// 以下为客户端全局连接断开回调函数
void conn_lost(void *context, char *cause)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
int rc;
printf("\nConnection lost\n");
if (cause) {
printf(" cause: %s\n", cause);
}
printf("Reconnecting\n");
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
conn_opts.maxRetryInterval = 16;
conn_opts.minRetryInterval = 1;
conn_opts.automaticReconnect = 1;
conn_opts.onFailure = onConnectFailure;
MQTTAsync_setConnected(client, client, conn_established);
if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start connect, return code %d\n", rc);
finished = 1;
}
}
int main(int argc, char* argv[])
{
// 创建异步连接客户端需要使用的属性结构体
MQTTAsync client;
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
int rc;
int ch;
// 创建异步连接客户端，不使用 Paho SDK 内置的持久化来处理缓存消息
if ((rc = MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL))
!= MQTTASYNC_SUCCESS)
{
printf("Failed to create client, return code %d\n", rc);
rc = EXIT_FAILURE;
goto exit;
}
// 设置异步连接回调，注意此处设置的回调函数为连接层面的全局回调函数
// conn_lost 为连接断开触发，有且只有连接成功后断开才会触发，在断开连接的情况下进行重连失败不触发。
// msgarrvd 收到消息时触发的回调函数
// msgdeliverd 是消息成功发送的回调函数，一般设置为NULL
if ((rc = MQTTAsync_setCallbacks(client, client, conn_lost, msgarrvd, msgdeliverd)) != MQTTASYNC_SUCCESS)
{
printf("Failed to set callbacks, return code %d\n", rc);
rc = EXIT_FAILURE;
goto destroy_exit;
}
// 设置连接参数
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
// 此处设置 API调用失败会触发的回调，接下来进行connect操作所以设置为 onConnectFailure 方法
conn_opts.onFailure = onConnectFailure;
// 此处设置客户端连接API调用成功会触发的回调，由于例程使用异步连接的 API，设置了会导致2个回调都被触发，所以建议不使用此回调
//conn_opts.onSuccess = onConnect;
// 注意第一次发起连接失败不会触发自动重连，只有曾经成功连接并断开后才会触发
conn_opts.automaticReconnect = 1;
//开启自动重连，并且设置 2-16s 的随机退避时间
conn_opts.maxRetryInterval = 16;
conn_opts.minRetryInterval = 2;
conn_opts.context = client;
// 设置异步回调函数，此与之前的 API 回调不同，每次连接/断开都会触发
MQTTAsync_setConnected(client, client, conn_established);
MQTTAsync_setDisconnected(client, client, disconnect_lost);
// 启动客户端连接，之前设置的 API 回调只会在这一次操作生效
if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start connect, return code %d\n", rc);
rc = EXIT_FAILURE;
goto destroy_exit;
}
......
}

复制代码