为什么线程越多服务反而越慢？从缓存行、调度到 TCP 消息边界

发表于 分类于 ， 阅读次数：
Linux 与 C++ 工程化 / 04更新于 2026-07-13预计阅读 9 分钟

关键词：CPU Cache、缓存行、MESI、进程、线程、IPC、调度、数据竞争、TCP Socket

一个并发服务器在单线程时运行正常。为了利用多核，你增加了工作线程，结果却出现三种看似无关的异常：两个线程明明修改不同的计数器，吞吐反而下降；共享总数偶尔少几个；客户端连续发送两条请求，服务端有时一次读到两条，有时只读到半条。

这些现象分别暴露了三个不同层次的边界：

1
2
3
硬件缓存行  → 决定“不同变量”是否仍会争用同一份缓存
C++ 内存模型 → 决定线程间的读写是否有定义、何时可见
TCP 字节流   → 保证字节可靠有序，但不保存应用消息边界

把三层混在一起，常会得到似是而非的解释，例如“MESI 会保证 ++counter 安全”或者“一次 write() 对应一次 read()”。本文从这台变慢又出错的服务器出发，用可运行的 C++ 示例说明每层到底保证什么、没有保证什么，以及工程代码应在哪里补上同步和协议。

缓存与线程示例使用 C++20；进程和 Socket 示例使用 POSIX API，可在 Linux 和 macOS 上编译，但 Windows 需要改用对应的进程与 Winsock API。缓存行大小、缓存层级和测量结果依赖具体 CPU；性能示例应使用优化构建、多次运行，并在相同负载下比较，不能把一次计时当成普遍结论。

1. 两个线程修改不同变量，为什么仍会互相拖慢？

CPU 的计算速度远快于内存访问速度。
如果每次读写数据都直接访问内存，CPU 大量时间会浪费在等待数据上。

所以现代 CPU 采用多级缓存结构：

1
2
寄存器 -> L1 Cache -> L2 Cache -> L3 Cache -> 内存 -> 硬盘/SSD
最快      很快        较快        较大        慢        更慢

常见结构：

  • L1 Cache：每个 CPU 核心私有，容量小，速度最快，通常分为指令缓存和数据缓存
  • L2 Cache：通常也是每个核心私有，容量比 L1 大，速度比 L1 慢
  • L3 Cache：通常多个核心共享，容量更大，速度更慢
  • 内存：所有核心共享，但访问延迟明显高于缓存

缓存的核心思想是局部性：

  • 时间局部性：刚访问过的数据，接下来可能还会访问
  • 空间局部性：访问某个地址后，附近地址也可能很快被访问

1.1 Cache Line

CPU 缓存不是按单个 int 或单个字节加载，而是按缓存行加载。
常见缓存行大小是 64 bytes。

例如读取一个 int，CPU 可能会把它周围连续的 64 字节一起加载进缓存。

1
2
3
内存地址：
0x1000 0x1004 0x1008 ... 0x103c
|------------- 一个 Cache Line -------------|

这也是为什么连续数组访问通常比链表访问更快：

  • 数组内存连续，缓存命中率高
  • 链表节点分散，容易频繁 cache miss

1.2 缓存加载过程

CPU 访问内存地址时，大致会把地址拆成：

  • tag：判断是不是目标内存块
  • index：定位缓存组
  • offset：定位缓存行内部偏移

简化理解：

1
2
3
4
5
6
CPU 读地址 A
-> 查 L1 是否有 A 所在的 Cache Line
-> L1 miss，则查 L2
-> L2 miss，则查 L3
-> L3 miss，则访问内存
-> 数据加载到缓存后，CPU 再读取

如果数据已经在缓存中，就是 cache hit。
如果数据不在缓存中，就需要向更低层级加载，就是 cache miss。

1.3 写策略：写穿与写回

CPU 修改数据时，不可能每次都立刻写回内存，否则缓存优势会被削弱。

常见写策略：

  • write-through：写缓存时同时写内存，简单但慢
  • write-back：先写缓存，并标记为 dirty，之后再批量写回内存

write-back 依赖脏位：

1
2
3
4
CPU 修改缓存行
-> 缓存行标记 dirty
-> 后续缓存行被替换或需要同步时
-> 再写回内存

1.4 多核缓存一致性：MESI

多核 CPU 下，每个核心都有自己的 L1/L2。
如果多个核心缓存了同一个变量，就会出现一致性问题。

例如：

1
2
3
4
Core 1 的 L1 里有 x = 1
Core 2 的 L1 里也有 x = 1
Core 1 把 x 改成 2
Core 2 继续读自己的旧缓存，可能还看到 x = 1

为了解决这个问题，CPU 使用缓存一致性协议。常见的是 MESI：

  • M Modified：当前核心独占并修改过，内存中是旧值
  • E Exclusive：当前核心独占，和内存一致
  • S Shared：多个核心共享，和内存一致
  • I Invalid：缓存行失效，不能再直接读取

当一个核心要写共享缓存行时，其他核心对应缓存行会被置为 Invalid。
其他核心之后再读，就必须重新加载新值。

1.5 代码示例：连续访问比跳跃访问更友好

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
#include <chrono>
#include <iostream>
#include <vector>

int main() {
    constexpr int N = 2048;
    std::vector<int> matrix(N * N, 1);

    auto now = std::chrono::steady_clock::now;

    long long sum1 = 0;
    auto t1 = now();
    for (int i = 0; i < N; ++i) {
        for (int j = 0; j < N; ++j) {
            sum1 += matrix[i * N + j]; // 连续访问
        }
    }
    auto t2 = now();

    long long sum2 = 0;
    for (int j = 0; j < N; ++j) {
        for (int i = 0; i < N; ++i) {
            sum2 += matrix[i * N + j]; // 跨行跳跃访问
        }
    }
    auto t3 = now();

    std::cout << "row-major sum = " << sum1 << ", time = "
              << std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1).count()
              << " ms\n";

    std::cout << "column-major sum = " << sum2 << ", time = "
              << std::chrono::duration_cast<std::chrono::milliseconds>(t3 - t2).count()
              << " ms\n";
}

二维数组在内存中通常按行连续存储。
按行访问能更好利用缓存行，按列访问容易产生更多 cache miss。

1.6 代码示例：伪共享 false sharing

伪共享指的是：两个线程修改不同变量，但变量刚好在同一个 Cache Line 中。
虽然逻辑上没有共享同一个变量，但硬件上会互相让对方的缓存行失效。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
#include <atomic>
#include <chrono>
#include <iostream>
#include <string_view>
#include <thread>

struct BadCounter {
    std::atomic<long long> a{0};
    std::atomic<long long> b{0};
};

struct GoodCounter {
    alignas(64) std::atomic<long long> a{0};
    alignas(64) std::atomic<long long> b{0};
};

template <class Counter>
void run(std::string_view label) {
    Counter counter;
    const auto start = std::chrono::steady_clock::now();

    std::thread t1([&] {
        for (int i = 0; i < 10'000'000; ++i) {
            counter.a.fetch_add(1, std::memory_order_relaxed);
        }
    });

    std::thread t2([&] {
        for (int i = 0; i < 10'000'000; ++i) {
            counter.b.fetch_add(1, std::memory_order_relaxed);
        }
    });

    t1.join();
    t2.join();

    const auto elapsed = std::chrono::steady_clock::now() - start;
    std::cout << label << ": "
              << std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count()
              << " ms, values="
              << counter.a.load(std::memory_order_relaxed) << ','
              << counter.b.load(std::memory_order_relaxed) << '\n';
}

int main() {
    run<BadCounter>("same cache line");
    run<GoodCounter>("separate cache lines");
}

编译时应开启优化，并重复运行：

1
2
clang++ -std=c++20 -O2 -pthread false_sharing.cpp -o false_sharing
./false_sharing

alignas(64) 的目的不是让变量更“高级”，而是尽量让两个热点变量落在不同缓存行，减少缓存一致性开销。64 字节只是常见硬件的取值，并非 C++ 对所有平台的保证；可靠的性能判断还需要结合目标 CPU 的缓存行信息和硬件计数器。memory_order_relaxed 在这里只承担原子计数，不负责发布其他数据。

2. 硬件已经有 MESI，为什么 ++counter 仍会出错？

2.1 进程是什么

进程是操作系统进行资源分配的基本单位。
一个进程通常拥有：

  • 独立虚拟地址空间
  • 代码段、数据段、堆、栈
  • 文件描述符表
  • 当前工作目录、环境变量
  • 信号处理方式

进程之间默认内存隔离。
一个进程崩溃，通常不会直接破坏另一个进程的地址空间。

2.2 线程是什么

线程是 CPU 调度执行的基本单位。
同一个进程内的多个线程共享：

  • 进程地址空间
  • 全局变量
  • 堆内存
  • 文件描述符

每个线程独有：

  • 栈
  • 寄存器上下文
  • 线程局部存储

所以线程通信方便，但也更容易出现数据竞争。

2.3 进程与线程对比

对比项 进程 线程
资源隔离 强 弱，同进程内共享内存
创建/切换成本 较高 较低
通信方式 IPC 共享内存、锁、条件变量
崩溃影响 通常影响自身 可能导致整个进程崩溃
典型场景 多服务、多 worker 隔离 并发任务、线程池、异步处理

2.4 进程通信 IPC

常见进程通信方式：

  • 管道 pipe：父子进程之间单向字节流通信
  • 命名管道 FIFO：无亲缘关系进程也可以通信
  • 信号 signal：事件通知，携带信息少
  • 共享内存 shm/mmap：最快，但需要同步机制
  • 消息队列：按消息传递
  • Unix Domain Socket：本机进程间通信，接口类似 socket
  • TCP/UDP Socket：跨机器通信，也可用于本机

2.5 代码示例：父子进程用 pipe 通信

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>

#include <cstring>
#include <iostream>

int main() {
    int fd[2];
    if (pipe(fd) == -1) {
        perror("pipe");
        return 1;
    }

    pid_t pid = fork();
    if (pid == -1) {
        perror("fork");
        return 1;
    }

    if (pid == 0) {
        close(fd[1]); // 子进程只读

        char buffer[128]{};
        ssize_t n = read(fd[0], buffer, sizeof(buffer) - 1);
        if (n > 0) {
            std::cout << "child read: " << buffer << '\n';
        }

        close(fd[0]);
        return 0;
    }

    close(fd[0]); // 父进程只写

    const char* msg = "hello from parent";
    write(fd[1], msg, std::strlen(msg));
    close(fd[1]);

    waitpid(pid, nullptr, 0);
}

管道本质是内核维护的一段缓冲区。
父进程写入，子进程读取，双方不直接共享用户态内存。

2.6 线程通信

线程在同一个进程内，最直接的通信方式是共享变量。
但共享变量必须解决同步问题。

常见同步手段：

  • std::mutex：互斥访问临界区
  • std::condition_variable：等待某个条件成立
  • std::atomic：无锁原子变量
  • 线程安全队列：生产者消费者模型

2.7 代码示例：线程用条件变量通信

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>

int main() {
    std::mutex mutex;
    std::condition_variable cv;
    std::queue<int> queue;
    bool done = false;

    std::thread producer([&] {
        for (int i = 0; i < 5; ++i) {
            {
                std::lock_guard<std::mutex> lock(mutex);
                queue.push(i);
            }
            cv.notify_one();
        }

        {
            std::lock_guard<std::mutex> lock(mutex);
            done = true;
        }
        cv.notify_one();
    });

    std::thread consumer([&] {
        while (true) {
            std::unique_lock<std::mutex> lock(mutex);
            cv.wait(lock, [&] {
                return done || !queue.empty();
            });

            if (queue.empty() && done) {
                break;
            }

            int value = queue.front();
            queue.pop();
            lock.unlock();

            std::cout << "consume: " << value << '\n';
        }
    });

    producer.join();
    consumer.join();
}

注意点：

  • wait 要使用谓词版本，防止虚假唤醒
  • 修改共享状态时要持有锁
  • 状态变化后用 notify_one 或 notify_all 唤醒等待线程

2.8 调度

操作系统调度器决定哪个线程或进程获得 CPU 时间。

常见状态：

1
2
3
4
就绪 Ready  -> 等待 CPU
运行 Running -> 正在 CPU 上执行
阻塞 Blocked -> 等待 IO、锁、条件变量、网络数据等
终止 Terminated -> 执行结束

调度器会考虑：

  • 优先级
  • 时间片
  • 是否阻塞
  • CPU 亲和性
  • 负载均衡
  • 交互性和吞吐量

Linux 普通任务使用公平调度类，但具体算法要结合内核版本理解：较老内核主要使用 CFS（Completely Fair Scheduler）；Linux 6.6 开始向 EEVDF（Earliest Eligible Virtual Deadline First）过渡。两者都不是简单轮流执行，而是依据虚拟运行时间、权重等信息分配 CPU；EEVDF 还通过 eligibility 与虚拟 deadline 改善延迟响应。排查调度行为时应先确认生产内核版本，不要把某一代算法的内部细节当成永恒接口。

2.9 代码示例：主动让出 CPU 与睡眠

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#include <chrono>
#include <iostream>
#include <thread>

int main() {
    std::thread worker([] {
        for (int i = 0; i < 5; ++i) {
            std::cout << "worker: " << i << '\n';

            // 提示调度器：当前线程愿意让出 CPU。
            std::this_thread::yield();

            // 当前线程进入睡眠，至少 10ms 内不会继续占用 CPU。
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    });

    worker.join();
}

yield() 只是建议调度器切换，不保证一定马上切走。
sleep_for() 会让线程进入阻塞/睡眠状态，到时间后重新进入就绪队列。

2.10 竞争问题

并发程序常见问题：

  • 数据竞争：多个线程同时访问同一变量，至少一个是写，且没有同步
  • 竞态条件：程序结果依赖不确定的执行顺序
  • 死锁：多个线程互相等待对方持有的资源
  • 饥饿：某个线程长期得不到执行机会或锁
  • 优先级反转：低优先级任务持有锁，高优先级任务反而被阻塞
  • 锁竞争：线程频繁争抢同一把锁，导致性能下降

2.11 代码示例：数据竞争与修复

错误写法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#include <iostream>
#include <thread>

int counter = 0;

void add() {
    for (int i = 0; i < 100000; ++i) {
        ++counter; // 数据竞争
    }
}

int main() {
    std::thread t1(add);
    std::thread t2(add);

    t1.join();
    t2.join();

    std::cout << counter << '\n'; // 结果不一定是 200000
}

用互斥锁修复：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
#include <iostream>
#include <mutex>
#include <thread>

int counter = 0;
std::mutex mutex;

void add() {
    for (int i = 0; i < 100000; ++i) {
        std::lock_guard<std::mutex> lock(mutex);
        ++counter;
    }
}

int main() {
    std::thread t1(add);
    std::thread t2(add);

    t1.join();
    t2.join();

    std::cout << counter << '\n';
}

如果只是简单计数，也可以用原子变量：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#include <atomic>
#include <iostream>
#include <thread>

std::atomic<int> counter{0};

void add() {
    for (int i = 0; i < 100000; ++i) {
        counter.fetch_add(1, std::memory_order_relaxed);
    }
}

int main() {
    std::thread t1(add);
    std::thread t2(add);

    t1.join();
    t2.join();

    std::cout << counter << '\n';
}

memory_order_relaxed 只保证自增本身是原子的，不保证额外的同步顺序。
用于纯计数通常可以，但不能随便替代锁。

3. 客户端写了两次，服务端为什么不能按两次读取？

Socket 是应用程序使用网络能力的系统调用接口。
TCP Socket 提供可靠、有序、面向连接的字节流。

3.1 TCP 服务端流程

典型服务端流程：

1
2
3
4
5
6
socket()
-> bind()
-> listen()
-> accept()
-> read()/write()
-> close()

含义：

  • socket：创建 socket 文件描述符
  • bind：绑定本机 IP 和端口
  • listen：进入监听状态
  • accept：从已完成连接队列中取出一个客户端连接
  • read/write：收发数据
  • close：关闭连接

3.2 TCP 客户端流程

典型客户端流程：

1
2
3
4
socket()
-> connect()
-> write()/read()
-> close()

connect() 会触发 TCP 三次握手。

3.3 TCP 三次握手

1
2
3
4
Client                         Server
  | -------- SYN ------------> |
  | <---- SYN + ACK ---------- |
  | -------- ACK ------------> |

目的：

  • 确认双方都能发送和接收
  • 协商初始序列号
  • 建立连接状态

三次握手完成后，服务端 accept() 才能返回一个已连接 socket。

3.4 TCP 四次挥手

1
2
3
4
5
Client                         Server
  | -------- FIN ------------> |
  | <-------- ACK ------------ |
  | <-------- FIN ------------ |
  | -------- ACK ------------> |

TCP 是全双工连接。
一方关闭发送方向，不代表另一方也立刻关闭，所以通常需要四次挥手。

3.5 监听 socket 与连接 socket

服务端有两类 socket：

  • 监听 socket：由 socket/bind/listen 创建，只负责接收新连接
  • 连接 socket：由 accept 返回，负责和某个客户端收发数据

示意：

1
2
3
4
5
listen_fd
  |
  | accept()
  v
conn_fd  <---->  client_fd

一个服务端通常只有少量监听 socket，但会有很多连接 socket。

3.6 代码示例：最小 TCP Echo 服务端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

#include <cerrno>
#include <cstring>
#include <iostream>

bool write_all(int fd, const void* data, std::size_t size) {
    const auto* p = static_cast<const char*>(data);
    while (size > 0) {
        const ssize_t n = write(fd, p, size);
        if (n < 0 && errno == EINTR) {
            continue;
        }
        if (n <= 0) {
            return false;
        }
        p += n;
        size -= static_cast<std::size_t>(n);
    }
    return true;
}

int main() {
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (listen_fd == -1) {
        perror("socket");
        return 1;
    }

    int reuse = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));

    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = htons(8080);

    if (bind(listen_fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) == -1) {
        perror("bind");
        close(listen_fd);
        return 1;
    }

    if (listen(listen_fd, SOMAXCONN) == -1) {
        perror("listen");
        close(listen_fd);
        return 1;
    }

    std::cout << "listen on 0.0.0.0:8080\n";

    while (true) {
        sockaddr_in client_addr{};
        socklen_t client_len = sizeof(client_addr);

        int conn_fd = accept(
            listen_fd,
            reinterpret_cast<sockaddr*>(&client_addr),
            &client_len
        );
        if (conn_fd == -1) {
            perror("accept");
            continue;
        }

        char buffer[1024];
        while (true) {
            ssize_t n = read(conn_fd, buffer, sizeof(buffer));
            if (n == 0) {
                break; // 客户端关闭连接
            }
            if (n < 0 && errno == EINTR) {
                continue;
            }
            if (n < 0) {
                perror("read");
                break;
            }

            if (!write_all(conn_fd, buffer, static_cast<std::size_t>(n))) {
                perror("write");
                break;
            }
        }

        close(conn_fd);
    }

    close(listen_fd);
}

这个服务端是阻塞、单连接串行模型，适合观察 socket → bind → listen → accept → read/write 的完整路径，不适合直接用于并发生产服务。write_all() 循环处理部分写，read() 则只消费当前已经到达的字节；二者都可能被信号打断。向已经关闭的连接写入还可能触发 SIGPIPE，正式服务需要结合平台选择忽略/捕获信号，或使用 MSG_NOSIGNAL、SO_NOSIGPIPE 等机制。

编译运行：

1
2
clang++ -std=c++20 server.cpp -o server
./server

测试：

1
nc 127.0.0.1 8080

3.7 代码示例：最小 TCP 客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

#include <cerrno>
#include <iostream>
#include <string>

bool transfer_exactly(int fd, void* data, std::size_t size, bool writing) {
    auto* p = static_cast<char*>(data);
    while (size > 0) {
        const ssize_t n = writing ? write(fd, p, size) : read(fd, p, size);
        if (n < 0 && errno == EINTR) {
            continue;
        }
        if (n <= 0) {
            return false;
        }
        p += n;
        size -= static_cast<std::size_t>(n);
    }
    return true;
}

int main() {
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd == -1) {
        perror("socket");
        return 1;
    }

    sockaddr_in server{};
    server.sin_family = AF_INET;
    server.sin_port = htons(8080);

    if (inet_pton(AF_INET, "127.0.0.1", &server.sin_addr) != 1) {
        std::cerr << "invalid address\n";
        close(fd);
        return 1;
    }

    if (connect(fd, reinterpret_cast<sockaddr*>(&server), sizeof(server)) == -1) {
        perror("connect");
        close(fd);
        return 1;
    }

    std::string msg = "hello socket\n";
    const std::size_t message_size = msg.size();
    if (!transfer_exactly(
            fd,
            msg.data(),
            message_size,
            true)) {
        perror("write");
        close(fd);
        return 1;
    }

    std::string response(message_size, '\0');
    if (!transfer_exactly(fd, response.data(), response.size(), false)) {
        std::cerr << "connection closed before the full echo arrived\n";
        close(fd);
        return 1;
    }
    std::cout << "server echo: " << response;

    close(fd);
}

客户端知道本次 Echo 应返回多少字节，因此可以循环读满固定长度。一般业务协议并不知道“这次要读多少”，必须使用下一节的分隔符或长度头。这里的辅助函数仅适用于阻塞 fd；非阻塞 Socket 遇到 EAGAIN/EWOULDBLOCK 时，应保存已完成进度并等待下一次可读或可写事件，不能在原地忙循环。

3.8 TCP 字节流与粘包

TCP 是字节流协议，不保留应用层消息边界。

例如客户端连续发送：

1
2
"hello"
"world"

服务端可能读到：

1
"helloworld"

也可能读到：

1
2
3
"he"
"llowor"
"ld"

所以应用层需要自己设计协议边界。常见方式：

  • 固定长度
  • 分隔符，例如 \n
  • 长度字段，例如前 4 字节表示 body 长度

3.9 代码示例：长度头协议

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
#include <arpa/inet.h>
#include <cerrno>
#include <cstdint>
#include <string>
#include <unistd.h>

constexpr std::uint32_t kMaxMessageSize = 1024 * 1024;

bool read_n(int fd, void* data, size_t n) {
    char* p = static_cast<char*>(data);
    size_t left = n;

    while (left > 0) {
        ssize_t nr = read(fd, p, left);
        if (nr < 0 && errno == EINTR) {
            continue;
        }
        if (nr <= 0) {
            return false;
        }
        p += nr;
        left -= static_cast<size_t>(nr);
    }

    return true;
}

bool write_n(int fd, const void* data, size_t n) {
    const char* p = static_cast<const char*>(data);
    size_t left = n;

    while (left > 0) {
        ssize_t nw = write(fd, p, left);
        if (nw < 0 && errno == EINTR) {
            continue;
        }
        if (nw <= 0) {
            return false;
        }
        p += nw;
        left -= static_cast<size_t>(nw);
    }

    return true;
}

bool send_message(int fd, const std::string& msg) {
    if (msg.size() > kMaxMessageSize) {
        return false;
    }

    std::uint32_t len = htonl(static_cast<std::uint32_t>(msg.size()));
    return write_n(fd, &len, sizeof(len)) &&
           write_n(fd, msg.data(), msg.size());
}

bool recv_message(int fd, std::string& msg) {
    std::uint32_t net_len = 0;
    if (!read_n(fd, &net_len, sizeof(net_len))) {
        return false;
    }

    const std::uint32_t len = ntohl(net_len);
    if (len > kMaxMessageSize) {
        return false;
    }
    msg.resize(len);

    return read_n(fd, msg.data(), msg.size());
}

这个例子先读固定 4 字节长度头，再按长度循环读取消息体，因此一次系统调用返回多少字节都不会改变协议边界。kMaxMessageSize 必须在 resize() 之前检查，否则对端只需发送一个伪造的大长度，就可能诱导服务分配巨量内存。

示例仍是阻塞模型：连接停在半个消息上时，线程会继续等待。真实服务还应设置超时，并区分“正常 EOF”“协议错误”“暂时不可读”和“系统错误”；非阻塞模式则要把已读取的 header/body 长度保存在连接状态中，等待后续事件继续解析。

4. 回到开头：线程越多，为什么可能更慢、更错？

开头的三种异常现在可以分别定位：两个独立原子计数器落在同一缓存行，会因为一致性流量产生伪共享；普通 ++counter 没有同步会形成 C++ 数据竞争，MESI 不能替语言层修复未定义行为；连续两次发送也不会形成两条 TCP 消息，应用必须自己定义长度或分隔符。

最重要的结论可以压缩为四点：

  1. 缓存一致性保证硬件缓存最终协调，不保证一组 C++ 操作具有业务原子性。
  2. 线程共享地址空间，因此通信成本低，但对象所有权、同步和退出条件都由程序负责。
  3. 进程提供更强隔离，代价是通过管道、共享内存或 Socket 等 IPC 显式交换数据。
  4. TCP 保证可靠、有序的字节流，不保证一次发送对应一次接收；read()、write() 的返回值必须驱动后续状态。

实践中不要一看到吞吐不足就先增加线程。先用 profiler 和硬件计数器确认热点，再检查共享数据布局与同步；网络侧则从第一天就把最大消息长度、部分读写、EOF、超时和错误状态写进协议处理逻辑。

三层边界可以这样记：

1
2
3
CPU Cache 影响单机内存访问性能
进程/线程 影响程序并发组织方式
Socket 影响进程之间、机器之间的通信方式

参考资料