1
2
3
4
5
6
7
8
9

buffer_ (std::vector<char>，初始大小 = 2048 + 8)

索引：  0    8                    tail_             buffer_.size()
        │    │                      │                    │
        ▼    ▼                      ▼                    ▼
        [prepend区][─── 可读数据 ───][──── 可写空间 ────]
        └── 8字节 ┘
             ↑                      ↑
           head_                  tail_

1
2
3
4
5
6
7
8
9

// MsgBuffer.cc 第 31 行
static constexpr size_t kBufferOffset{8};  // prepend 预留大小

// 构造函数（MsgBuffer.cc 第 34-37 行）
MsgBuffer::MsgBuffer(size_t len)
    : head_(kBufferOffset),    // head_ 从 8 开始
      initCap_(len),           // 记录初始容量，用于 retrieveAll 的缩容
      buffer_(len + head_),    // 总容量 = 用户要求 + 8字节 prepend
      tail_(head_)             // tail_ 也从 8 开始，初始可读字节为 0

1
2
3
4

总大小: 2056 字节
head_ = tail_ = 8
可读字节: 0
可写字节: 2048

1
2

peek()   ── 只看，不消费（不移动 head_）
read()   ── 读取并消费（移动 head_）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// 正确的网络包解析流程
while (buf.readableBytes() >= sizeof(PacketHeader)) {
    // 1. 先 peek 解析包头，检查包是否完整
    auto header = *(PacketHeader*)buf.peek();
    uint32_t packetLen = buf.peekInt32();  // 自动做大端转换

    if (buf.readableBytes() < packetLen) break; // 包还没收全，等下一次

    // 2. 包完整了，才 retrieve 消费掉
    buf.retrieve(packetLen);
}

1
2
3
4
5

// 写入时自动转换为网络字节序（大端）
buf.appendInt32(0x12345678);   // 实际写入: 12 34 56 78

// 读取时自动转换回主机字节序
uint32_t val = buf.peekInt32();  // 内部调用 ntohl()

1
2
3
4
5

场景：发送一个自定义协议包
  ┌───────┬─────────────────┐
  │ 4字节  │    包体数据      │
  │ 包长度 │  (先写入这部分)  │
  └───────┴─────────────────┘

1
2
3
4
5
6

MsgBuffer buf;
// 1. 先写包体（可能很复杂）
buf.append(payloadData, payloadLen);

// 2. 再在头部插入包长度（利用 prepend 区，不需要移动 payloadData）
buf.addInFrontInt32(buf.readableBytes());

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

情况1: head_ >= len（prepend 区够用）
  → 直接写入 head_ - len 处，head_ -= len
  → O(1)，零拷贝！

情况2: head_ 不够，但 writableBytes() >= len
  → 把可读数据往后挪，腾出头部空间
  → O(n) memmove，但无需分配内存

情况3: 完全不够
  → 创建新 buffer，拷贝所有数据
  → O(n)，最慢

1
2
3
4
5
6
7
8
9

操作前：
[  8字节 prepend  ][  可读数据  ][  可写  ]
 ↑               ↑ head_=8
 0

addInFrontInt32(packetLen)：
[4字节][4字节][  可读数据  ][  可写  ]
       ↑ head_=4
写入包长度

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

void MsgBuffer::retrieve(size_t len) {
    if (len >= readableBytes()) { retrieveAll(); return; }
    head_ += len;  // 只移动指针，O(1)，不移动数据
}

void MsgBuffer::retrieveAll() {
    // 如果 buffer 因扩容变得很大，此时趁机缩容回初始大小
    if (buffer_.size() > (initCap_ * 2)) {
        buffer_.resize(initCap_ + kBufferOffset);
    }
    tail_ = head_ = kBufferOffset;  // 重置指针
}

 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

// MsgBuffer.cc 第 148-173 行
ssize_t MsgBuffer::readFd(int fd, int *retErrno)
{
    char extBuffer[8192];   // 栈上 8 KB 临时缓冲区
    struct iovec vec[2];

    size_t writable = writableBytes();
    vec[0].iov_base = begin() + tail_;  // 第一块：buffer 的可写区
    vec[0].iov_len  = writable;
    vec[1].iov_base = extBuffer;         // 第二块：栈上临时区
    vec[1].iov_len  = sizeof(extBuffer);

    // 如果 buffer 可写空间 >= 8192，只用一块（不需要 extBuffer）
    const int iovcnt = (writable < sizeof extBuffer) ? 2 : 1;
    ssize_t n = ::readv(fd, vec, iovcnt);  // 一次系统调用读入两块

    if (n < 0) {
        *retErrno = errno;
    } else if (static_cast<size_t>(n) <= writable) {
        // 数据全在 buffer 里，直接移动 tail_
        tail_ += n;
    } else {
        // 数据溢出到了 extBuffer，把 extBuffer 的部分 append 进来
        tail_ = buffer_.size();
        append(extBuffer, n - writable);  // append 内部会 ensureWritableBytes
    }
    return n;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

socket 内核缓冲区有 10KB 数据，buffer 可写空间只有 3KB

readv 调用：
  vec[0] → buffer 可写区 3KB
  vec[1] → 栈上 extBuffer 8KB

读取结果 n = 10KB：
  vec[0] 填满：3KB → 直接 tail_ = buffer_.size()
  vec[1] 填充：7KB → append(extBuffer, 7KB) → 触发 ensureWritableBytes 扩容

最终：buffer 扩容，所有 10KB 数据都在 buffer 里

 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

// MsgBuffer.cc 第 39-60 行
void MsgBuffer::ensureWritableBytes(size_t len)
{
    if (writableBytes() >= len) return;  // 够用，直接返回

    // 策略1：整理碎片（head_ 前面有空间 + 尾部可写 >= len）
    if (head_ + writableBytes() >= (len + kBufferOffset))
    {
        // 把可读数据向前移动到 kBufferOffset 处
        std::copy(begin() + head_, begin() + tail_, begin() + kBufferOffset);
        tail_ = kBufferOffset + (tail_ - head_);
        head_ = kBufferOffset;
        return;  // 不分配新内存！
    }

    // 策略2：真正扩容
    size_t newLen;
    if ((buffer_.size() * 2) > (kBufferOffset + readableBytes() + len))
        newLen = buffer_.size() * 2;  // 双倍扩容（避免频繁 realloc）
    else
        newLen = readableBytes() + len;  // 刚好够用
    MsgBuffer newbuffer(newLen);
    newbuffer.append(*this);
    swap(newbuffer);  // 交换指针，O(1)
}

1
2
3
4
5
6
7
8

需要写入 len 字节
      │
      ├─ 尾部可写 >= len ──────────────► 直接写，什么都不做
      │
      ├─ 头部空闲 + 尾部可写 >= len+8 ── memmove 整理碎片，无需分配内存
      │    （数据消费后 head_ 往右移，头部积累了空闲空间）
      │
      └─ 真的不够 ─────────────────────► 创建新 buffer，2倍扩容

1
2

[  大量空闲  ][少量可读数据][小量可写]
 ↑            ↑head_      ↑tail_

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

// BufferNode.h 第 31-84 行
class BufferNode : public NonCopyable {
public:
    // 多态识别（默认 false，文件节点重写为 true）
    virtual bool isFile()   const { return false; }
    virtual bool isStream() const { return false; }
    virtual bool isAsync()  const { return false; }

    // 核心接口：获取当前可发送的数据指针和长度
    virtual void getData(const char *&data, size_t &len) = 0;

    // 标记已发送了 len 字节（移动内部读指针）
    virtual void retrieve(size_t len) = 0;

    // 还剩多少字节没发
    virtual long long remainingBytes() const = 0;

    // 仅内存节点支持
    virtual void append(const char *, size_t) { LOG_FATAL << "Not a memory buffer node"; }

    // 仅文件节点支持（用于 sendfile）
    virtual int getFd() const { LOG_FATAL << "Not a file buffer node"; return -1; }
};

1
2
3
4
5
6
7

TcpConnection 的发送队列：
┌──────────────────────────────────────────────────┐
│  std::list<BufferNodePtr> sendNodeList_           │
│                                                  │
│  [MemBufferNode]─►[FileBufferNode]─►[MemBufferNode]│
│   "HTTP 头部"        "静态文件内容"    "trailing data"│
└──────────────────────────────────────────────────┘

 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

// 假设自定义协议：
// [4字节 包总长度][2字节 消息ID][N字节 消息体]

void onReceiveData(const TcpConnectionPtr &conn, MsgBuffer *buf)
{
    while (buf->readableBytes() >= 6)  // 最小包头 6 字节
    {
        // 1. peek 包长度（不消费）
        uint32_t packetLen = buf->peekInt32();

        // 2. 检查完整性
        if (buf->readableBytes() < packetLen) break;

        // 3. 读消息 ID
        buf->retrieve(4);  // 跳过包长度字段
        uint16_t msgId = buf->readInt16();  // 读并消费 2 字节

        // 4. 读消息体
        size_t bodyLen = packetLen - 6;
        std::string body = buf->read(bodyLen);

        // 5. 分发处理
        dispatchMessage(conn, msgId, body);
    }
}

// 构造响应包（在头部插入长度）
MsgBuffer buildResponse(uint16_t msgId, const std::string &body)
{
    MsgBuffer buf;
    buf.appendInt16(msgId);
    buf.append(body.c_str(), body.size());
    // 在头部插入总长度（利用 prepend 区）
    buf.addInFrontInt32(buf.readableBytes() + 4);
    return buf;
}

1
2
3
4

void MsgBuffer::retrieve(size_t len) {
    if (len >= readableBytes()) { retrieveAll(); return; }
    head_ += len;  // 只移动指针，旧数据仍残留在 [旧head_, 新head_) 区间
}

1
2
3
4

vec[0].iov_len = writable;          // buffer 可写区大小
vec[1].iov_len = sizeof(extBuffer); // 8192 字节
const int iovcnt = (writable < sizeof extBuffer) ? 2 : 1;
ssize_t n = ::readv(fd, vec, iovcnt);

1
2
3
4
5
6
7

// 进入情况2的前提：head_ < len（prepend区不够），但 writableBytes() >= len
if (len <= writableBytes()) {
    std::copy(begin() + head_, begin() + tail_, begin() + head_ + len);
    memcpy(begin() + head_, buf, len);
    tail_ += len;
    return;
}

1
2
3
4

情况2的前提条件：head_ < len，且尾部可写 >= len

[  head_字节  ][── 可读数据 ──][──── 可写空间(>= len) ────]
0            head_            tail_                    size()

1
2
3
4
5
6
7

操作前：
[小空间][─── 可读数据 ───][──── 足够的可写空间 ────]
       ↑head_            ↑tail_

操作后（右移 len 字节）：
[小空间][新数据 len字节][─── 可读数据 ───][剩余可写]
       ↑head_                            ↑tail_(+=len)

1
2
3
4

// MsgBuffer.h 第 190-193 行
void appendInt8(const uint8_t b) {
    append(static_cast<const char *>((void *)&b), 1);  // 直接写入，无需转换
}

1
2
3
4
5

// MsgBuffer.cc 第 80-84 行
void MsgBuffer::appendInt16(const uint16_t s) {
    uint16_t ss = htons(s);  // 主机序 → 网络序（大端）
    append(static_cast<const char *>((void *)&ss), 2);
}