关于物联网:MQTT-客户端自动重连最佳实践｜构建可靠-IoT-设备连接

背景

MQTT 是一个基于 TCP 协定的公布 / 订阅模型协定，它被广泛应用于物联网、传感器网络和其余低带宽、不稳固网络环境中。在这些网络环境中，网络连接往往不稳固，可能会呈现网络故障、信号弱化、丢包等问题，这可能会导致 MQTT 客户端与服务器之间的连贯中断。物联网利用中，常见的触发断线重连的场景包含：

1. 网络环境恶劣或者断网，造成 MQTT 客户端连贯超时断开。
2. 因为业务须要服务端降级切换，服务端被动敞开断开。
3. 设施重启或客户端重启，客户端被动重连。
4. 其余网络因素造成 TCP/IP 传输层断开导致 MQTT 连贯重连。

为了确保 MQTT 客户端与服务器之间的稳固连贯，MQTT 客户端须要实现重连逻辑，帮忙 MQTT 客户端主动从新连贯服务器，并复原之前的订阅关系、放弃会话等状态。

为什么 MQTT 客户端重连代码须要良好的设计

MQTT 设施重连是很多物联网利用中不可避免的状况。设计 MQTT 客户端重连逻辑时须要留神应用正确的事件回调办法，每次重连设置正当的随机退却工夫，以保障客户端和服务端的长时间稳固运行，从而确保业务的失常发展。

不合理的重连逻辑设计可能会造成诸多问题：

1. 重连逻辑生效导致客户端静默不再承受 Broker 音讯。
2. 客户端频繁重连，无重连退却工夫导致造成 DDOS 攻打服务端 Broker。
3. 客户端频繁高低线导致 Broker 服务端资源适量不必要的耗费。

而正当的重连逻辑既能够进步 MQTT 客户端的稳定性和可靠性，防止因网络连接中断而导致的数据失落、提早等问题，还能够升高因为频繁连贯对服务器端的压力。

如何设计一段 MQTT 客户端重连代码

在进行 MQTT 客户端重连代码设计时须要思考以下几个方面：

• 设置正确的连贯保活工夫 MQTT 客户端的连贯保活工夫即 Keep Alive，负责检测以后连贯的衰弱状态。Keep Alive 超时会触发客户端重连和服务端敞开客户端连贯。该数值会影响到服务端和客户端检测到连贯断开不可用的时长，用户须要依据本身网络状态，以及冀望的最长等待时间来设置正当的 Keep Alive。
• 重连策略和退却 用户应该依据网络环境的不同，制订不同的重连策略。例如，当网络连接中断时，能够设置一个初始等待时间，并在每次重连尝试后逐步减少等待时间，以防止网络连接中断导致的大量重连尝试。倡议应用指数退却算法或随机 + 阶梯延时来留出足够的退却时隙。
• 连贯状态治理 须要在客户端中保护连贯状态，包含连贯状态的记录、连贯断开的起因、已订阅的主题列表等信息。当连贯中断时，客户端应该记录下连贯断开的起因，并进行相应的重连尝试。但如果应用会话放弃性能，则不须要客户端本人保留这些信息。
• 异样解决 在连贯过程中可能会产生各种异常情况，例如服务器不可用、认证失败、网络异样等。须要在客户端中增加异样解决逻辑，依据异常情况进行相应的解决。MQTT 5 协定提供了详实的此类断开连接起因，客户端能够依据这些信息记录异样日志、断开连接、再次重连等。
• 最大尝试次数限度 对于一些低功耗设施，为防止重连次数过多导致客户端资源耗费过大，有时候须要思考限度最大重连尝试次数。当超过最大尝试次数后，客户端应该停止重连尝试进入休眠状态，防止无意义的重连。
• 退却算法 有两种罕用的重连退却办法：指数弥补算法和随机退却。指数弥补算法是通过负反馈机制指数减少等待时间来找到适合的发送 / 连贯速率。随机退却即通过设置等待时间的上上限，每次重连都期待随机的延时工夫，因为其易于实现而有宽泛应用。

重连代码示例

咱们将以 Paho MQTT C 的库为例，示范如何应用异步编程模型优雅实现主动重连性能。Paho 提供了丰盛的回调函数，请留神不同回调办法触发条件和设置形式不同，别离有全局回调、API 回调和异步办法回调。API 回调有相当的灵活性，但当开启主动重连性能时，倡议只应用异步回调。此处对三种回调函数都提供了例程，用户能够应用此例程验证三种回调函数的触发。

// 是 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;
    }

    ......
}

查看 MQTTAsync_subscribe.c 具体代码。

更多抉择：NanoSDK 内置重连策略

NanoSDK 是除了 Paho 以外的又一 MQTT SDK 抉择。NanoSDK 基于 NNG-NanoMSG 我的项目开发，应用 MIT License，对开源和商业都很敌对。相较于 Paho 其最大的不同在于内置的全异步 I/O 和 反对 Actor 编程模型，当应用 QoS 1/2 音讯时能够取得更高的音讯吞吐速率。而且 NanoSDK 反对 MQTT over QUIC 协定，与大规模物联网 MQTT 音讯服务器 EMQX 5.0 联合可解决弱网下的数据传输难题。这些劣势使得它曾经在车联网和工业场景中失去了宽泛的应用。

在 NanoSDK 中，重连策略曾经齐全内置，无需用户手动实现。

// nanosdk 采纳主动拨号机制，默认进行重连
nng_dialer_set_ptr(*dialer, NNG_OPT_MQTT_CONNMSG, connmsg);
nng_dialer_start(*dialer, NNG_FLAG_NONBLOCK);

总结

本文介绍在 MQTT 客户端代码实现过程中，重连逻辑设计的重要性与最佳实际。通过本文，读者能够设计更为正当的 MQTT 设施重连代码，升高客户端与服务器端的资源开销，构建更加稳固牢靠的物联网设施连贯。

版权申明：本文为 EMQ 原创，转载请注明出处。

原文链接：https://www.emqx.com/zh/blog/mqtt-client-auto-reconnect-best-practices