📚 课程目录第4章 MQTT协议核心概念

4. MQTT协议核心概念

TCP 是网络通信的基石。MQTT 和 HTTP 都跑在 TCP 之上，所以搞懂 TCP 客户端和服务器的实现，是后面所有实战的前提。我个人习惯把 TCP 通信比作打电话——先拨号（连接），再说话（收发数据），最后挂断（关闭）。这个比喻虽然简单，但核心流程一点不差。

4.1 TCP 通信流程：一张图说清楚

先看整体流程。服务端和客户端各自走一套固定的步骤，谁先谁后，哪步能阻塞，哪步能超时，心里要有数。

这张图我建议你存下来。每次写 TCP 代码前扫一眼，能避免很多低级错误。我在项目中见过有人把 bind() 和 listen() 的顺序搞反，结果 socket 一直报错——说白了就是流程没走对。

4.2 Echo 服务器实现：最简单的 TCP 服务

Echo 服务器是 TCP 编程的 "Hello World"。它做什么？客户端发什么，它就原样返回什么。听起来简单，但麻雀虽小五脏俱全。

下面是一个基于 Zephyr 的 Echo 服务器实现。注意看每一步的 POSIX API 调用，Zephyr 的网络栈对 BSD Socket 做了兼容，所以代码风格和 Linux 下几乎一样。

#include <zephyr/kernel.h>
#include <zephyr/net/socket.h>
#include <zephyr/net/net_if.h>
#include <stdio.h>
#include <string.h>

#define PORT 8888
#define BUFFER_SIZE 256

void echo_server(void)
{
    int server_fd, client_fd;
    struct sockaddr_in server_addr, client_addr;
    socklen_t client_len = sizeof(client_addr);
    char buffer[BUFFER_SIZE];
    int ret;

    // 1. 创建 socket
    server_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (server_fd < 0) {
        printk("socket() failed\n");
        return;
    }

    // 2. 绑定地址
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = INADDR_ANY;
    server_addr.sin_port = htons(PORT);

    ret = bind(server_fd, (struct sockaddr *)&server_addr,
               sizeof(server_addr));
    if (ret < 0) {
        printk("bind() failed\n");
        close(server_fd);
        return;
    }

    // 3. 开始监听
    ret = listen(server_fd, 5);
    if (ret < 0) {
        printk("listen() failed\n");
        close(server_fd);
        return;
    }

    printk("Echo server listening on port %d\n", PORT);

    while (1) {
        // 4. 接受客户端连接
        client_fd = accept(server_fd,
                           (struct sockaddr *)&client_addr,
                           &client_len);
        if (client_fd < 0) {
            printk("accept() failed\n");
            continue;
        }

        printk("Client connected\n");

        // 5. 收发数据
        while (1) {
            ret = recv(client_fd, buffer, sizeof(buffer) - 1, 0);
            if (ret <= 0) {
                // 连接断开或出错
                break;
            }
            buffer[ret] = '\0';
            printk("Received: %s\n", buffer);

            // 原样返回
            ret = send(client_fd, buffer, ret, 0);
            if (ret < 0) {
                printk("send() failed\n");
                break;
            }
        }

        // 6. 关闭客户端连接
        close(client_fd);
        printk("Client disconnected\n");
    }

    close(server_fd);
}

我的经验： 第一次跑这个代码时，记得检查 Zephyr 的 prj.conf 里有没有打开网络功能。我踩过这个坑——代码编译通过，但板子死活连不上，最后发现是 CONFIG_NETWORKING 没开。你需要在 prj.conf 里至少加上这几行：

CONFIG_NETWORKING=y
CONFIG_NET_TCP=y
CONFIG_INIT_STACKS=y

4.3 TCP 客户端实现：主动发起连接

客户端比服务端简单。它不需要 bind() 和 listen()，直接 socket() 然后 connect() 就行。connect() 会触发 TCP 三次握手，握手成功就建立连接。

#include <zephyr/kernel.h>
#include <zephyr/net/socket.h>
#include <stdio.h>
#include <string.h>

#define SERVER_IP "192.168.1.100"
#define SERVER_PORT 8888
#define BUFFER_SIZE 256

void tcp_client(void)
{
    int sock;
    struct sockaddr_in server_addr;
    char buffer[BUFFER_SIZE];
    int ret;

    // 1. 创建 socket
    sock = socket(AF_INET, SOCK_STREAM, 0);
    if (sock < 0) {
        printk("socket() failed\n");
        return;
    }

    // 2. 配置服务器地址
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(SERVER_PORT);
    inet_pton(AF_INET, SERVER_IP, &server_addr.sin_addr);

    // 3. 连接服务器
    ret = connect(sock, (struct sockaddr *)&server_addr,
                  sizeof(server_addr));
    if (ret < 0) {
        printk("connect() failed\n");
        close(sock);
        return;
    }

    printk("Connected to server\n");

    // 4. 发送数据
    strcpy(buffer, "Hello from Zephyr!");
    ret = send(sock, buffer, strlen(buffer), 0);
    if (ret < 0) {
        printk("send() failed\n");
        close(sock);
        return;
    }

    // 5. 接收回显
    ret = recv(sock, buffer, sizeof(buffer) - 1, 0);
    if (ret > 0) {
        buffer[ret] = '\0';
        printk("Echo: %s\n", buffer);
    }

    // 6. 关闭连接
    close(sock);
}

注意： connect() 默认是阻塞的。如果服务器没启动或者网络不通，你的线程会一直卡在 connect() 上。我曾经在一个产品里遇到过——设备上电后连不上服务器，结果整个任务挂死，其他功能全停了。解决办法是给 connect() 设置超时，或者用非阻塞模式。

4.4 多客户端处理：从单线程到 poll/select

上面的 Echo 服务器有个致命问题：一次只能处理一个客户端。当 accept() 返回后，服务器就卡在 recv() 里，第二个客户端连不上来。这在实际项目中根本不能用。

解决思路有两个：多线程或多路复用。在嵌入式环境里，多线程的开销往往太大，所以 poll() 和 select() 是更常见的选择。

4.4.1 poll() 的核心思想

poll() 的本质是：你把一堆文件描述符交给内核，告诉它「帮我盯着，哪个有数据可读、哪个可写、哪个出错了」，然后你阻塞等待。内核一旦发现有事件发生，就唤醒你，你再去处理对应的 socket。

这样做的好处是——一个线程就能管理几十上百个连接。你想想看，每个连接一个线程，100 个连接就是 100 个线程，光栈空间就吃掉几 MB 的 RAM，在 Zephyr 上根本扛不住。

4.4.2 使用 poll() 的多客户端 Echo 服务器

#include <zephyr/kernel.h>
#include <zephyr/net/socket.h>
#include <stdio.h>
#include <string.h>

#define PORT 8888
#define MAX_CLIENTS 10
#define BUFFER_SIZE 256

void echo_server_poll(void)
{
    int server_fd, ret, i;
    struct sockaddr_in server_addr;
    struct pollfd fds[MAX_CLIENTS + 1];
    int nfds = 1;
    char buffer[BUFFER_SIZE];

    // 创建监听 socket
    server_fd = socket(AF_INET, SOCK_STREAM, 0);
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = INADDR_ANY;
    server_addr.sin_port = htons(PORT);

    bind(server_fd, (struct sockaddr *)&server_addr,
         sizeof(server_addr));
    listen(server_fd, 5);

    // 初始化 pollfd 数组
    memset(fds, 0, sizeof(fds));
    fds[0].fd = server_fd;
    fds[0].events = POLLIN;

    printk("Poll-based echo server started\n");

    while (1) {
        ret = poll(fds, nfds, -1);  // 永久等待
        if (ret < 0) {
            printk("poll() failed\n");
            continue;
        }

        // 检查监听 socket 是否有新连接
        if (fds[0].revents & POLLIN) {
            int client_fd = accept(server_fd, NULL, NULL);
            if (client_fd >= 0 && nfds < MAX_CLIENTS + 1) {
                fds[nfds].fd = client_fd;
                fds[nfds].events = POLLIN;
                nfds++;
                printk("New client, total: %d\n", nfds - 1);
            }
        }

        // 检查每个客户端是否有数据
        for (i = 1; i < nfds; i++) {
            if (fds[i].revents & POLLIN) {
                ret = recv(fds[i].fd, buffer, sizeof(buffer) - 1, 0);
                if (ret <= 0) {
                    // 客户端断开
                    close(fds[i].fd);
                    fds[i].fd = -1;
                    printk("Client disconnected\n");
                } else {
                    buffer[ret] = '\0';
                    send(fds[i].fd, buffer, ret, 0);
                }
            }
        }

        // 清理已断开的客户端
        for (i = 1; i < nfds; i++) {
            if (fds[i].fd == -1) {
                for (int j = i; j < nfds - 1; j++) {
                    fds[j] = fds[j + 1];
                }
                nfds--;
                i--;
            }
        }
    }
}

  关键点总结：
  pollfd 数组的第一个元素永远是监听 socket，只关注 POLLIN（有新连接）
客户端 socket 也只关注 POLLIN（有数据到达）
每次 poll() 返回后，要遍历所有 fd，检查 revents 字段
客户端断开后，要把对应的 fd 从数组中移除，否则下次 poll() 会一直返回 POLLIN

4.4.3 poll() vs select()：怎么选？

对比项	poll()	select()
最大 fd 数量	无硬限制（受内存限制）	受 FD_SETSIZE 限制（通常 1024）
参数结构	pollfd 数组，清晰易扩展	fd_set 位图，操作繁琐
性能	O(n) 遍历，但每次只需传递数组	O(n) 遍历，且每次要重建 fd_set
Zephyr 支持	✅ 完全支持	✅ 完全支持
我的推荐	✅ 优先使用	兼容旧代码时使用

我个人习惯用 poll()。原因很简单——select() 的 fd_set 有大小限制，而且每次调用前都要用 FD_SET/FD_ZERO 重新构造，代码写起来啰嗦。poll() 的数组结构更直观，扩展性也更好。

避坑指南： 我曾经在 poll() 里犯过一个低级错误——每次 poll() 返回后，我只处理了 fds[0] 的新连接，忘了遍历后面的客户端 fd。结果新客户端能连上来，但旧客户端发数据服务器没反应。排查了半天才发现是循环条件写错了。所以写 poll() 循环时，一定要把「处理新连接」和「处理已有连接」分开，逻辑才清晰。

4.5 本章小结

TCP 编程的核心就三件事：建立连接、收发数据、关闭连接。Echo 服务器是练手的最佳起点，poll() 是嵌入式多客户端处理的标配方案。你把这些基础打牢了，后面学 MQTT 和 HTTP 时就会发现——它们本质上都是在 TCP 之上封装了一层应用协议，底层的 socket 操作万变不离其宗。