《Java 网络编程基础》7.2 半包、粘包与缓冲区增长

00 05 68 65 6c 6c 6f 00 05 77 6f 72 6c 64

情况 A：一次 read 拿到全部 14 字节  -> 两个完整帧
情况 B：第一次拿到 00 05 68 65      -> 半个帧

package com.plume.linkd.core;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;

public final class AccumulatingReader implements Handler {

    @FunctionalInterface
    public interface FrameHandler {
        void onFrame(Connection c, byte[] frame);
    }

    public static final int DEFAULT_MAX_BUFFERED = 1 << 20;

    private final BufferPool pool;
    private final FrameDecoder decoder;
    private final FrameHandler frames;
    private final int maxBuffered;
    private ByteBuffer acc;

    public AccumulatingReader(BufferPool pool, FrameDecoder decoder, FrameHandler frames) {
        this(pool, decoder, frames, DEFAULT_MAX_BUFFERED);
    }

    public AccumulatingReader(BufferPool pool, FrameDecoder decoder,
                              FrameHandler frames, int maxBuffered) {
        if (maxBuffered < 64) {
            throw new IllegalArgumentException("maxBuffered too small: " + maxBuffered);
        }
        this.pool = pool;
        this.decoder = decoder;
        this.frames = frames;
        this.maxBuffered = maxBuffered;
        this.acc = pool.acquire(1024);
    }

    @Override
    public void onReadable(Connection c) {
        try {
            int n;
            do {
                if (!acc.hasRemaining()) {
                    grow();
                }
                n = c.channel().read(acc);
            } while (n > 0);
            drainFrames(c);
            if (n == -1) {
                c.close();
            }
        } catch (IOException e) {
            c.close();
        }
    }

    @Override
    public void onWritable(Connection c) {
        // 本类只处理入站数据，出站由 Connection 的写队列负责。
    }

    @Override
    public void onClosed(Connection c) {
        if (acc != null) {
            pool.release(acc);
            acc = null;
        }
    }

    public int capacity() {
        return acc.capacity();
    }

    public int bufferedBytes() {
        return acc.position();
    }

    private void drainFrames(Connection c) {
        acc.flip();
        List<byte[]> decoded = decoder.decode(acc);
        acc.compact();
        for (byte[] frame : decoded) {
            frames.onFrame(c, frame);
        }
    }

    private void grow() throws IOException {
        int current = acc.capacity();
        int next = current >= maxBuffered / 2 ? maxBuffered : current * 2;
        if (next <= current) {
            throw new IOException("frame exceeds maxBuffered=" + maxBuffered);
        }
        ByteBuffer bigger = pool.acquire(next);
        acc.flip();
        bigger.put(acc);
        pool.release(acc);
        acc = bigger;
    }
}

chunk 1 (00) -> buffered=1, cap=1024
chunk 2 (05 he) -> buffered=4, cap=1024
frame=hello
chunk 3 (llo 00) -> buffered=1, cap=1024
buffered before last byte = 1, cap=1024
frame=world
chunk 4 (05 world) -> buffered=0, cap=1024
big frame bytes = 3002, cap before = 1024
frame=xxxx…（3000 个 x，长行截断）
cap after big frame = 4096, buffered=0

package com.plume.linkd.core;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.ArrayDeque;
import java.util.Deque;

public final class WriteQueue {

    private final Deque<ByteBuffer> queue = new ArrayDeque<>();
    private final BufferPool pool;
    private final long highWaterMark;
    private long pending;

    public WriteQueue(BufferPool pool, long highWaterMark) {
        this.pool = pool;
        this.highWaterMark = highWaterMark;
    }

    public boolean enqueue(ByteBuffer out) {
        int size = out.remaining();
        if (pending + size > highWaterMark) {
            return false;
        }
        queue.addLast(out);
        pending += size;
        return true;
    }

    public boolean writeTo(SocketChannel ch) throws IOException {
        while (!queue.isEmpty()) {
            ByteBuffer head = queue.peekFirst();
            int n = ch.write(head);
            if (n < 0) {
                throw new IOException("channel closed while writing");
            }
            pending -= n;
            if (head.hasRemaining()) {
                return false;
            }
            queue.pollFirst();
            pool.release(head);
        }
        return true;
    }

    public boolean isEmpty() {
        return queue.isEmpty();
    }

    public long pendingBytes() {
        return pending;
    }

    public int queuedBuffers() {
        return queue.size();
    }

    public void clear() {
        ByteBuffer buf;
        while ((buf = queue.pollFirst()) != null) {
            pool.release(buf);
        }
        pending = 0;
    }
}

enqueue 64KiB = true
drained=false, pending=64512, buffers=1

AccumulatingReader reader = new AccumulatingReader(pool, decoder, (conn, frame) -> {
    ByteBuffer out = pool.acquire(frame.length);
    out.put(frame).flip();
    conn.enqueue(out);
});
loop.register(ch, reader);

// 反例：position 在数据末尾，decode 看不到任何字节
List<byte[]> frames = decoder.decode(acc);

// 反例：每来一块就复制整个累积缓冲区
byte[] copy = new byte[acc.position() + n];
System.arraycopy(acc.array(), 0, copy, 0, acc.position());

// 反例：没写出去的字节被清掉
int n = ch.write(buf);
if (n == 0) {
    buf.clear();
}