28、物联网网络编程：MQTT协议、CoAP协议、MQTT服务器实现、传感器数据采集

物联网（IoT）编程，说白了就是让设备跟设备、设备跟服务器“说话”。我做了这么多年网络编程，发现很多新手一上来就搞HTTP，结果在低带宽、弱网环境下吃尽苦头。物联网场景下，MQTT和CoAP才是真正的主角。今天咱们就聊聊这两个协议，再手把手搭一个轻量级MQTT服务器，把传感器数据采上来。

28.1 MQTT协议：轻量级消息传输

MQTT（Message Queuing Telemetry Transport）是IBM开发的发布/订阅模式协议。它设计得极其精简，最小报文只有2个字节。我在一个智能农业项目里用过它，传感器节点用电池供电，跑MQTT能撑一年多不换电池——HTTP根本做不到。

核心概念

  • Broker（代理）：消息中转站，所有客户端都连它
  • Topic（主题）：消息分类标签，支持层级结构（如 sensor/temp/room1）
  • Publish（发布）：发送消息到某个主题
  • Subscribe（订阅）：接收感兴趣的主题消息
  • QoS（服务质量）：0（最多一次）、1（至少一次）、2（恰好一次）

重要提醒：QoS 0 最快但可能丢消息，QoS 2 最可靠但开销大。我建议传感器上报用 QoS 1，控制指令用 QoS 2。

MQTT报文结构

一个MQTT控制报文包含固定头、可变头和有效载荷。固定头第一个字节的高4位表示报文类型，低4位是标志位。举个例子，CONNECT报文的固定头是 0x10。

// MQTT CONNECT报文示例（十六进制）
10 0E 00 04 4D 51 54 54 04 02 00 3C 00 04 64 65 6D 6F
|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  +-- 客户端ID长度(4)
|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  +-- 客户端ID("demo")
|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  +-- 保持连接时间(60秒)
|  |  |  |  |  |  |  |  |  |  |  |  |  |  +-- 标志位(0x02)
|  |  |  |  |  |  |  |  |  |  |  |  |  +-- 协议版本(4=MQTT 3.1.1)
|  |  |  |  |  |  |  |  |  |  |  |  +-- 协议名长度(4)
|  |  |  |  |  |  |  |  |  |  |  +-- 协议名("MQTT")
|  |  |  |  |  |  |  |  |  |  +-- 剩余长度(14)
|  |  |  |  |  |  |  |  |  +-- 固定头(0x10=CONNECT)

28.2 CoAP协议：类RESTful的UDP协议

CoAP（Constrained Application Protocol）是专门为资源受限设备设计的。它基于UDP，使用类似HTTP的GET/POST/PUT/DELETE方法。我记得有个智能路灯项目，每个节点只有2KB RAM，跑CoAP刚刚好，MQTT都嫌重。

CoAP vs MQTT

特性 MQTT CoAP
传输层 TCP UDP
模型 发布/订阅 请求/响应
报文开销 2字节起 4字节起
可靠性 QoS机制 CON/NON消息
适用场景 双向通信、数据上报 设备控制、资源查询

我的经验：如果设备需要主动上报数据，选MQTT；如果是服务器查询设备状态，选CoAP。两个协议可以互补使用。

28.3 实现一个轻量级MQTT服务器

市面上有Mosquitto、EMQX这些成熟Broker，但有时候我们需要定制化。比如我做过一个边缘网关项目，需要在嵌入式Linux上跑一个极简Broker。下面是我用C语言实现的核心框架。

服务器架构

先看整体结构，我用SVG画了一张图，帮你理解数据流。

传感器节点 MQTT Client 执行器节点 MQTT Client 网关设备 MQTT Client MQTT Broker 主题树管理 会话管理 消息转发 QoS处理 云平台 数据存储 远程监控 ▲ 传感器数据上报 → Broker转发 → 云平台存储

核心代码实现

下面是一个简化版Broker的核心循环。它监听1883端口，处理CONNECT、PUBLISH、SUBSCRIBE三种报文。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <pthread.h>

#define MQTT_PORT 1883
#define MAX_CLIENTS 64
#define TOPIC_MAX 128

// 客户端结构体
typedef struct {
    int fd;
    char client_id[64];
    char topics[TOPIC_MAX][64];
    int topic_count;
    int connected;
} mqtt_client;

mqtt_client clients[MAX_CLIENTS];
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

// 解析MQTT固定头
int parse_fixed_header(unsigned char byte, int *type, int *flags) {
    *type = (byte >> 4) & 0x0F;
    *flags = byte & 0x0F;
    return 0;
}

// 处理CONNECT报文
void handle_connect(int client_index, unsigned char *buffer, int len) {
    // 跳过固定头和剩余长度
    int offset = 2;
    // 协议名长度
    int proto_len = (buffer[offset] << 8) | buffer[offset+1];
    offset += 2 + proto_len;
    // 协议版本
    int version = buffer[offset++];
    // 连接标志
    int flags = buffer[offset++];
    // 保持连接时间
    int keepalive = (buffer[offset] << 8) | buffer[offset+1];
    offset += 2;
    
    // 读取客户端ID
    int id_len = (buffer[offset] << 8) | buffer[offset+1];
    offset += 2;
    memcpy(clients[client_index].client_id, &buffer[offset], id_len);
    clients[client_index].client_id[id_len] = '\0';
    clients[client_index].connected = 1;
    
    printf("[CONNECT] Client: %s, Keepalive: %d\n", 
           clients[client_index].client_id, keepalive);
    
    // 发送CONNACK
    unsigned char connack[] = {0x20, 0x02, 0x00, 0x00};
    write(clients[client_index].fd, connack, 4);
}

// 处理PUBLISH报文
void handle_publish(int client_index, unsigned char *buffer, int len) {
    int offset = 2; // 跳过固定头
    // 主题长度
    int topic_len = (buffer[offset] << 8) | buffer[offset+1];
    offset += 2;
    char topic[128];
    memcpy(topic, &buffer[offset], topic_len);
    topic[topic_len] = '\0';
    offset += topic_len;
    
    // 消息内容
    int msg_len = len - offset;
    char message[256];
    memcpy(message, &buffer[offset], msg_len);
    message[msg_len] = '\0';
    
    printf("[PUBLISH] Topic: %s, Message: %s\n", topic, message);
    
    // 转发给订阅了该主题的客户端
    pthread_mutex_lock(&lock);
    for (int i = 0; i < MAX_CLIENTS; i++) {
        if (i != client_index && clients[i].connected) {
            for (int j = 0; j < clients[i].topic_count; j++) {
                if (strcmp(clients[i].topics[j], topic) == 0) {
                    write(clients[i].fd, buffer, len);
                    break;
                }
            }
        }
    }
    pthread_mutex_unlock(&lock);
}

// 处理SUBSCRIBE报文
void handle_subscribe(int client_index, unsigned char *buffer, int len) {
    int offset = 2; // 跳过固定头
    // 报文标识符
    int pkt_id = (buffer[offset] << 8) | buffer[offset+1];
    offset += 2;
    
    // 读取主题过滤器
    int topic_len = (buffer[offset] << 8) | buffer[offset+1];
    offset += 2;
    char topic[128];
    memcpy(topic, &buffer[offset], topic_len);
    topic[topic_len] = '\0';
    offset += topic_len;
    int qos = buffer[offset] & 0x03;
    
    // 保存订阅
    pthread_mutex_lock(&lock);
    int idx = clients[client_index].topic_count;
    strcpy(clients[client_index].topics[idx], topic);
    clients[client_index].topic_count++;
    pthread_mutex_unlock(&lock);
    
    printf("[SUBSCRIBE] Client: %s, Topic: %s, QoS: %d\n",
           clients[client_index].client_id, topic, qos);
    
    // 发送SUBACK
    unsigned char suback[] = {0x90, 0x03, (pkt_id >> 8) & 0xFF, pkt_id & 0xFF, qos};
    write(clients[client_index].fd, suback, 5);
}

// 客户端线程处理函数
void *client_handler(void *arg) {
    int client_index = *(int *)arg;
    unsigned char buffer[1024];
    int n;
    
    while ((n = read(clients[client_index].fd, buffer, sizeof(buffer))) > 0) {
        int type, flags;
        parse_fixed_header(buffer[0], &type, &flags);
        
        switch (type) {
            case 1: // CONNECT
                handle_connect(client_index, buffer, n);
                break;
            case 3: // PUBLISH
                handle_publish(client_index, buffer, n);
                break;
            case 8: // SUBSCRIBE
                handle_subscribe(client_index, buffer, n);
                break;
            case 14: // DISCONNECT
                clients[client_index].connected = 0;
                close(clients[client_index].fd);
                printf("[DISCONNECT] Client: %s\n", clients[client_index].client_id);
                return NULL;
            default:
                printf("[UNKNOWN] Type: %d\n", type);
        }
    }
    
    clients[client_index].connected = 0;
    close(clients[client_index].fd);
    return NULL;
}

int main() {
    int server_fd;
    struct sockaddr_in addr;
    
    // 创建TCP socket
    server_fd = socket(AF_INET, SOCK_STREAM, 0);
    int opt = 1;
    setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
    
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = INADDR_ANY;
    addr.sin_port = htons(MQTT_PORT);
    
    bind(server_fd, (struct sockaddr *)&addr, sizeof(addr));
    listen(server_fd, MAX_CLIENTS);
    
    printf("MQTT Broker started on port %d\n", MQTT_PORT);
    
    while (1) {
        struct sockaddr_in client_addr;
        socklen_t addr_len = sizeof(client_addr);
        int client_fd = accept(server_fd, (struct sockaddr *)&client_addr, &addr_len);
        
        // 找个空位
        int idx = -1;
        for (int i = 0; i < MAX_CLIENTS; i++) {
            if (!clients[i].connected) {
                idx = i;
                break;
            }
        }
        
        if (idx >= 0) {
            clients[idx].fd = client_fd;
            clients[idx].topic_count = 0;
            clients[idx].connected = 0;
            
            pthread_t tid;
            pthread_create(&tid, NULL, client_handler, &idx);
            pthread_detach(tid);
        } else {
            printf("Max clients reached, rejecting connection\n");
            close(client_fd);
        }
    }
    
    close(server_fd);
    return 0;
}

注意：这个实现只处理了最基本的报文类型。生产环境还需要处理：遗嘱消息、保留消息、QoS 2的完整流程、心跳检测、主题通配符匹配等。我曾经在生产环境漏了心跳检测，结果一堆死连接占着资源，排查了半天。

28.4 传感器数据采集实战

假设我们有一个温度传感器，每隔5秒采集一次数据，通过MQTT上报。下面是一个完整的客户端示例。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <time.h>

// 模拟温度采集
float read_temperature() {
    // 实际项目中这里会调用硬件接口
    // 比如读取ADC值、I2C传感器等
    return 20.0 + (rand() % 100) / 10.0;
}

// 构建MQTT PUBLISH报文
int build_publish_packet(unsigned char *buffer, const char *topic, 
                         const char *message) {
    int idx = 0;
    int topic_len = strlen(topic);
    int msg_len = strlen(message);
    int remaining = 2 + topic_len + msg_len;
    
    // 固定头：PUBLISH (0x30) + QoS 0
    buffer[idx++] = 0x30;
    
    // 剩余长度编码
    if (remaining < 128) {
        buffer[idx++] = remaining;
    } else {
        buffer[idx++] = (remaining % 128) | 0x80;
        buffer[idx++] = remaining / 128;
    }
    
    // 主题长度
    buffer[idx++] = (topic_len >> 8) & 0xFF;
    buffer[idx++] = topic_len & 0xFF;
    
    // 主题
    memcpy(&buffer[idx], topic, topic_len);
    idx += topic_len;
    
    // 消息内容
    memcpy(&buffer[idx], message, msg_len);
    idx += msg_len;
    
    return idx;
}

int main() {
    int sock;
    struct sockaddr_in server_addr;
    unsigned char buffer[256];
    
    // 连接Broker
    sock = socket(AF_INET, SOCK_STREAM, 0);
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(1883);
    inet_pton(AF_INET, "127.0.0.1", &server_addr.sin_addr);
    
    if (connect(sock, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) {
        perror("connect");
        return -1;
    }
    
    // 发送CONNECT报文
    // 这里简化处理，直接发送一个预构建的CONNECT包
    unsigned char connect_pkt[] = {
        0x10, 0x0E, // 固定头
        0x00, 0x04, 'M', 'Q', 'T', 'T', // 协议名
        0x04,       // 协议版本
        0x02,       // 标志位
        0x00, 0x3C, // 保持连接60秒
        0x00, 0x04, 's', 'e', 'n', '1' // 客户端ID
    };
    write(sock, connect_pkt, sizeof(connect_pkt));
    
    // 读取CONNACK
    read(sock, buffer, 4);
    printf("Connected to MQTT Broker\n");
    
    // 循环采集并上报
    srand(time(NULL));
    while (1) {
        float temp = read_temperature();
        char message[32];
        snprintf(message, sizeof(message), "%.1f", temp);
        
        int pkt_len = build_publish_packet(buffer, "sensor/temp", message);
        write(sock, buffer, pkt_len);
        
        printf("Published: sensor/temp -> %s\n", message);
        sleep(5);
    }
    
    close(sock);
    return 0;
}

28.5 避坑指南与性能优化

做物联网项目，坑是真不少。我把自己踩过的几个大坑分享给你。

我曾经...在一个智能家居项目里，所有设备都用QoS 2上报数据。结果Broker内存暴涨，最后OOM被系统杀了。后来改成传感器用QoS 1，控制指令用QoS 2，问题就解决了。记住：不是所有消息都需要最高可靠性。

常见问题

  • 心跳超时：客户端断网后，Broker要等一个keepalive周期才能发现。我建议keepalive设短一点（比如10秒），配合LWT（遗嘱消息）及时通知其他设备。
  • 主题爆炸：每个设备都用独立主题，订阅端要维护大量过滤规则。更好的做法是用层级结构，比如 sensor/{room_id}/{type}，订阅端用通配符 sensor/+/temp。
  • 报文粘包：TCP是流协议，一次read可能读到多个MQTT报文。必须根据剩余长度字段正确切分。我见过有人直接按固定大小读，结果数据全乱了。
  • 内存碎片：嵌入式设备频繁malloc/free会导致碎片。建议用内存池或者预分配缓冲区。

性能优化建议

优化项 方法 效果
减少报文大小 使用短主题名、二进制编码 节省30%带宽
批量发布 合并多个传感器数据到一条消息 减少网络交互次数
会话持久化 启用Clean Session=0 断线重连不丢消息
线程池 用epoll+线程池替代每连接一线程 支持更多并发连接

我的习惯：在嵌入式设备上，我会把MQTT库静态编译，去掉不需要的功能（比如WebSocket、TLS），能省下几十KB的Flash空间。对于传感器数据，用Protobuf或MessagePack做序列化，比JSON省一半流量。

好了，MQTT和CoAP的核心内容就这些。物联网编程的关键是选对协议、控制开销、处理好异常。你想想看，一个温度传感器上报的数据丢了也就丢了，但一个阀门控制指令丢了可能造成事故——所以理解每个QoS级别的含义，比会写代码更重要。

公众号：蓝海资料掘金营，微信 deep3321

← 上一章 📖 返回目录 下一章 →