Azure IoT C SDK（MQTT）：已排队出站消息的超时替代方案咨询

1. 自定义本地消息队列+超时管控 #

• 在调用SDK发送接口前，自行维护本地消息队列，每条消息附加创建时间戳
• 定时轮询队列，自动剔除超过设定超时阈值的消息并释放内存
• 限制队列最大容量，达到上限时根据业务需求选择丢弃最旧消息或拒绝新消息
• 核心实现示例：

#define MAX_QUEUE_CAPACITY 100
#define MSG_EXPIRE_SEC 30

typedef struct {
    IOTHUB_MESSAGE_HANDLE msg_handle;
    time_t create_timestamp;
} LocalQueuedMsg;

LocalQueuedMsg msg_queue[MAX_QUEUE_CAPACITY];
int queue_head = 0, queue_tail = 0;

// 添加消息到本地队列
bool enqueue_msg(IOTHUB_MESSAGE_HANDLE msg) {
    int next_tail = (queue_tail + 1) % MAX_QUEUE_CAPACITY;
    if (next_tail == queue_head) {
        // 队列已满，丢弃最旧消息腾出空间
        IoTHubMessage_Destroy(msg_queue[queue_head].msg_handle);
        queue_head = (queue_head + 1) % MAX_QUEUE_CAPACITY;
    }
    msg_queue[queue_tail].msg_handle = msg;
    msg_queue[queue_tail].create_timestamp = time(NULL);
    queue_tail = next_tail;
    return true;
}

// 清理超时消息
void purge_expired_msgs() {
    time_t now = time(NULL);
    while (queue_head != queue_tail) {
        if (difftime(now, msg_queue[queue_head].create_timestamp) > MSG_EXPIRE_SEC) {
            IoTHubMessage_Destroy(msg_queue[queue_head].msg_handle);
            queue_head = (queue_head + 1) % MAX_QUEUE_CAPACITY;
        } else {
            break; // 队列按时间排序，后续消息不会超时
        }
    }
}

// 批量发送队列中有效消息
void process_local_queue(IOTHUB_DEVICE_CLIENT_HANDLE client) {
    purge_expired_msgs();
    while (queue_head != queue_tail) {
        IOTHUB_CLIENT_RESULT send_result = IoTHubDeviceClient_SendEventAsync(
            client, msg_queue[queue_head].msg_handle, send_complete_callback, msg_queue[queue_head].msg_handle);
        if (send_result == IOTHUB_CLIENT_OK) {
            IoTHubMessage_Destroy(msg_queue[queue_head].msg_handle);
            queue_head = (queue_head + 1) % MAX_QUEUE_CAPACITY;
        } else {
            // 发送失败，退出等待下次重试
            break;
        }
    }
}

2. 配置SDK重试策略+回调跟踪消息生命周期 #

• 利用SDK内置的重试策略，设置合理的重试次数和间隔，避免消息无限期重试滞留
• 通过发送完成回调，在消息多次发送失败时主动销毁释放内存
• 配置示例：

// 设置指数退避重试策略，最大重试5次
IOTHUB_CLIENT_RETRY_POLICY retry_policy = {
    .type = IOTHUB_CLIENT_RETRY_EXPONENTIAL_BACKOFF,
    .exponentialBackoff.retryCount = 5,
    .exponentialBackoff.minBackoff = 1000,
    .exponentialBackoff.maxBackoff = 30000
};
IoTHubDeviceClient_SetRetryPolicy(client, &retry_policy);

// 发送完成回调，处理失败消息
void send_complete_callback(IOTHUB_CLIENT_CONFIRMATION_RESULT result, void* context) {
    IOTHUB_MESSAGE_HANDLE msg = (IOTHUB_MESSAGE_HANDLE)context;
    if (result != IOTHUB_CLIENT_CONFIRMATION_OK) {
        // 重试耗尽仍失败，销毁消息释放内存
        IoTHubMessage_Destroy(msg);
    }
}

// 发送时传入消息作为上下文，方便回调处理
IoTHubDeviceClient_SendEventAsync(client, msg, send_complete_callback, msg);

3. 限制SDK内部消息缓存大小 #

• 部分SDK版本支持通过IoTHubDeviceClient_SetOption设置MAX_SEND_BUFFER_SIZE选项，限制内部缓存的待发送消息数量
• 示例代码：

size_t max_send_buffer = 50; // 最多缓存50条消息
IoTHubDeviceClient_SetOption(client, "MAX_SEND_BUFFER_SIZE", &max_send_buffer);

• 注意：该选项的支持性依赖SDK版本，需对照对应版本源码确认

4. 连接状态监听+主动清理队列 #

• 注册连接状态回调，当检测到设备与云端断开连接时，主动清理本地队列及SDK内部待发送消息
• 示例：

void connection_status_callback(IOTHUB_CLIENT_CONNECTION_STATUS status, IOTHUB_CLIENT_CONNECTION_STATUS_REASON reason, void* context) {
    IOTHUB_DEVICE_CLIENT_HANDLE client = (IOTHUB_DEVICE_CLIENT_HANDLE)context;
    if (status == IOTHUB_CLIENT_CONNECTION_DISCONNECTED) {
        // 清理本地超时消息
        purge_expired_msgs();
        // 取消SDK内部所有待发送请求（部分版本支持）
        IoTHubDeviceClient_CancelAllSends(client);
    }
}

IoTHubDeviceClient_SetConnectionStatusCallback(client, connection_status_callback, client);

内容的提问来源于stack exchange，提问作者Benjamin Fiset-Deschênes