未分类

MQTT 자동 재연결 모범 사례

무료무료 기술 자료

이 글은 바로 읽을 수 있으며 기초 학습과 검색 유입에 적합합니다.

modbus기술팀璀璨会员 약 20 분 읽기
MQTT 자동 재연결 모범 사례
理解和적용 MQTT 프로토콜是物联网领域中的一项重要技能。为了确保设备和服务器之间的稳定연결，我们需要深入了解并有效적용 MQTT 클라이언트的自动重连特性。下面，让我们像在探索一个神秘的冒险岛一样，深入探索 MQTT 프로토콜和重连机制。 MQTT 是一种基于 TCP 프로토콜的发布/订阅模型프로토콜。它像一艘经过风雨洗礼的海船，在物联网、센서네트워크和其他低带宽、不稳定네트워크환경中航行。但是，就像海上的风浪，네트워크환경中也充满了各种挑战：네트워크故障、信号弱化、데이터丢包等等，这些都可能使得 MQTT 클라이언트与服务器之间的연결中断。在物联网的大海中，常见的触发断线重连的风浪包括네트워크환경恶劣或断网、服务器업그레이드、设备或클라이언트重启、以及其他네트워크因素等。 在这种情况下，我们如何确保我们的 "船"（MQTT 클라이언트）始终能与 "港口"（服务器）保持稳定的연결呢？答案就是我们需要给我们的 "船" 装备一套自动导航系统——MQTT 클라이언트的自动重连逻辑。 设计一个优秀的 MQTT 클라이언트重连逻辑，就像建造一艘坚固的海船。如果设计得不合理，那么我们的 "船" 可能会失去导航，静默不再수신来自 "港口" 的消息，甚至可能会因为频繁地尝试重连而无意识地攻击我们的 "港口"，这就如同在海上无头乱窜，不仅消耗了自身的能量，也给 "港口" 带来了不必要的压力。然而，如果我们的重连逻辑设计得合理，那么无论何时失去연결，我们的 "船" 都能稳定地自动导航，重新找到 "港口"，并保持与其的연결。 设计 MQTT 클라이언트重连逻辑时，我们需要考虑几个关键因素：
  1. 航行保活时间：在 MQTT 中，我们称其为 Keep Alive。这是一个定时器，它会定期检查我们的 "船" 是否与 "港口" 保持연결。我们需要根据实际的네트워크환경和적용需求，来设置一个合适的 Keep Alive。
  2. 重连策略和退避：当我们的 "船" 失去了与 "港口" 的연결，我们不应立刻尝试重新연결，而应该设置一个合理的대기时间，以免过度消耗자료。这就像是当我们的船在海上迷路时，我们需要暂时停下，观察风向、测量海流，然后再制定新的航行路线。我们可以使用指数退避算法或者阶梯式的延时策略来实现这个功能。
  3. 연결状态管理：我们的 "船" 需要一个航海로그，来记录与 "港口" 的연결状态、연결断开的原因、已经订阅的信息等重要信息。在연결断开时，我们的 "船" 应该查阅航海로그，分析연결断开的原因，然后尝试重新연결 "港口"。
  4. 异常处理：在航行过程中，我们的 "船" 可能会遇到各种各样的문제，例如 "港口" 不可用、인증失败、네트워크异常等。我们的 "船" 需要有一个应急计划，来应对这些문제。例如，当 "港口" 不可用时，我们的 "船" 可能需要寻找其他的 "港口"；当인증失败时，我们的 "船" 可能需要检查自身的인증信息是否正确；当네트워크异常时，我们的 "船" 可能需要暂停航行，대기네트워크恢复正常。
  5. 最大尝试次数限制：对于一些低功耗设备，我们可能需要考虑限制尝试重连的次数，以避免过度消耗设备的电力。就像在海上迷航的 "船"，当它已经尝试了很多次都无法找到 "港口" 时，可能就需要暂时停下，대기更好的航行条件。
在设计了这个自动导航系统（MQTT 클라이언트的自动重连逻辑）之后，我们的 "船" 就能更好地在物联网的海洋中航行，无论面临何种挑战，都能始终保持与 "港口" 的稳定연결，从而确保我们的적용能够顺利进行。 来看一个实际的案例。我们以 Paho MQTT C 库为例，它为我们提供了一套丰富的航海도구——回调函数，让我们可以根据实际情况设定自动导航系统的工作방식。Paho 提供了全局回调、API 回调和异步方法回调，让我们可以在各种情况下都能保持与 "港口" 的연결。 这就是我们如何在物联网的海洋中航行的故事。希望通过这个故事，能够帮助你更好地理解 MQTT 프로토콜和重连机制，也希望你的 "船" 能在物联网的海洋中顺利航行。   
/*******************************************************************************
 * 복사right (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 주소     "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_success데이터* response);
void onConnectFailure(void* context, MQTTAsync_failure데이터* response);
void onSubscribe(void* context, MQTTAsync_success데이터* response);
void onSubscribeFailure(void* context, MQTTAsync_failure데이터* 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("订阅主题 %s\n使用클라이언트 %s 并用QoS%d\n\n"
           "按Q<Enter>退出\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("开始订阅失败，返回码 %d\n", 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("     原因: %s\n", 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("开始연결失败，返回码 %d\n", rc);
		finished = 1;
	}
}

// 收到消息时的全局回调函数，此处简单的打印消息
int msgarrvd(void *context, char *topic이름, int topicLen, MQTTAsync_message *message)
{
    printf("消息已到达\n");
    printf("     主题: %s\n", topic이름);
    printf("   消```C
息: ");

    /* 打印消息콘텐츠 */
    char* payloadptr = message->payload;
    for(int i = 0; i < message->payloadlen; i++)
    {
        putchar(*payloadptr++);
    }
    putchar('\n');

    /* 释放消息内存 */
    MQTTAsync_freeMessage(&message);
    MQTTAsync_free(topic이름);

    return 1;
}

/* 异步断开연결的回调函数 */
void onDisconnect(void* context, MQTTAsync_success데이터* response)
{
    printf("成功断开연결\n");
    disc_finished = 1;
}

/* 异步연결成功的回调函数，在연결成功的时候进行订阅操作。 */
void onConnect(void* context, MQTTAsync_success데이터* response)
{
    MQTTAsync client = (MQTTAsync)context;
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    int rc;

    printf("成功연결\n");

    printf("订阅主题 %s\n使用클라이언트 %s 并用QoS%d\n\n"
           "按Q<Enter>退出\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("开始订阅失败，返回码 %d\n", rc);
        finished = 1;
    }
}

/* 异步연결失败的回调函数 */
void onConnectFailure(void* context, MQTTAsync_failure데이터* response)
{
    printf("연결失败\n");
    if (response && response->message)
    {
        printf("失败信息: %s\n", response->message);
    }
    finished = 1;
}

/* 异步订阅成功的回调函数 */
void onSubscribe(void* context, MQTTAsync_success데이터* response)
{
    printf("成功订阅\n");
    subscribed = 1;
}

/* 异步订阅失败的回调函数 */
void onSubscribeFailure(void* context, MQTTAsync_failure데이터* response)
{
    printf("订阅失败\n");
    if (response && response->message)
    {
        printf("失败信息: %s\n", response->message);
    }
    finished = 1;
}

/* 异步취소订阅的回调函数 */
void onUnsubscribe(void* context, MQTTAsync_success데이터* 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, 주소, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);

    /* 设置全局回调函数 */
    MQTTAsync_set호명backs(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, M Q TT AS Y NC _ SU CC ESS)
    {
        print f (" 연 결 시작 실패 , 반환 코드 % d \ n ", r c);
        EX IT _ FA IL UR E 를 반환 합니다 .
    }

    While (아 직 도 !완료 됨)
    {
        # if defined (_ W IN 32)
            잠 들 기 (1 000)
        # else
            잠 들 다 (1)
        # 엔 디 프
    }

    if (sub s cribe)
    {
        if ((r c = M Q T TA syn c _ uns ub s cribe (cli ent , TOP IC , N ULL)) ! = M Q TT AS Y NC _ SU CC ESS)
        {
            print f (" 구 독 취소 실패 , 반환 코드 % d \ n ", r c);
            EX IT _ FA IL UR E 를 반환 합니다 .
        }
    }
    
    / * 연결 끊 기 */
    M Q T TA syn c _ dis conne ct Op tions disc _ op ts = M Q T TA syn c _ dis conne ct Op tions _ in itia li zer ;
    disc _ op ts . on S uc cess = on Dis conne ct ;
    if ((rc = MQTTAsync_disconnect(client, M Q TT AS Y NC _ SU CC ESS)
    {
        print f (" % d \ n " 반환 코드 에서 연결 해 제 시작 실패);
        EX IT _ FA IL UR E 를 반환 합니다 .
    }
    While (아 직 도 ! disc _ finished)
    {
        # if defined (_ W IN 32)
            잠 들 기 (1 000)
        # else
            잠 들 다 (1)
        # 엔 디 프
    }

    / * 클 라이언 트 제거 */
    MQTTAsync_destroy(

    EX IT _ SU CC ESS 를 반환 합니다 .
}
 
후원미니 포스터공유
AI 추천 글
相关标签
실제 프로젝트에 이 자료를 활용하시겠습니까?

도구 센터에서 프레임 분석, CRC 검증, 장비 디버깅을 진행하거나 요청을 제출하여 선정 및 연동 제안을 받아보세요.

엔지니어 회원

이 글을 실행 가능한 디버깅 자료로 활용하세요

가입하면 고급 프레임 분석, 자료 패키지 다운로드, 코드 예제, 엔지니어링 사례, 우선 기술 지원을 이용할 수 있어 실제 프로젝트 납품에 적합합니다.

고급 도구 무제한 이용
자료 패키지와 코드 패키지
완전한 엔지니어링 사례 라이브러리
우선 기술 지원 채널
회원 혜택 보기 먼저 디버깅 도구 사용
KEEP READING

관련 기술 자료

2026년 6월 28일 Modbus 在工业物联网中的进阶적용：边缘계산、MQTT 桥接与클라우드 플랫폼接入实战 2026년 6월 28일 Modbus 在主流 PLC 中的编程实战指南：西门子、三菱、台达、汇川全시리즈 2026년 6월 28일 Modbus CRC/LRC 체크섬원리与编程实现：从数学推导到多언어代码实战

답글 작성

이메일 주소는 공개되지 않습니다. 必填项已用 * 标注