netty接收缓冲区大小适配及类型选择

Channel创建

再次回顾一下Channel的流程吧。

doBind

public ChannelFuture bind(int inetPort) {
    return bind(new InetSocketAddress(inetPort));
}

bind2

public ChannelFuture bind(SocketAddress localAddress) {
    validate();
    if (localAddress == null) {
        throw new NullPointerException("localAddress");
    }
    return doBind(localAddress);
}

doBind

private ChannelFuture doBind(final SocketAddress localAddress) {
    final ChannelFuture regFuture = initAndRegister();
    ...
}

initAndRegister

final ChannelFuture initAndRegister() {
    ...
    channel = channelFactory.newChannel();
    ...
}

newChannel

默认采用的是反射进行实例化。

public T newChannel() {
    ...
    return clazz.newInstance();
}

NioServerSocketChannel

使用的就是传入的NioServerSocketChannel类，反射默认调用的是无参构造。

public NioServerSocketChannel() {
    this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}

newSocket

private static ServerSocketChannel newSocket(SelectorProvider provider) {
    ....
    return provider.openServerSocketChannel();
}

次方法返回的，就是原生的NIO的ServerSocketChannel了

NioServerSocketChannel2

实际调用的构造方法

public NioServerSocketChannel(ServerSocketChannel channel) {
    super(null, channel, SelectionKey.OP_ACCEPT);
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

此时已经传入OP_ACCEPT事件监听

Channel实例信息

NioServerSocketChannel

public NioServerSocketChannel(ServerSocketChannel channel) {
    super(null, channel, SelectionKey.OP_ACCEPT);
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

super

protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    super(parent, ch, readInterestOp);
}

super

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    super(parent);
    this.ch = ch;
    ...
}

可以看到，NIO的真实ServerSocketChannel作为内部对象，被封装了。

super

protected AbstractChannel(Channel parent) {
   this.parent = parent;
   id = newId();
   unsafe = newUnsafe();
   pipeline = newChannelPipeline();
}

NioServerSocket的id和pipeline此时被创建和添加，具体的创建办法可以继续跟踪。

非主要目的，先到此为止。

Channel配置信息

NioServerSocketChannel

public NioServerSocketChannel(ServerSocketChannel channel) {
    super(null, channel, SelectionKey.OP_ACCEPT);
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

config之前，先看看javaChannel().socket()

javaChannel

protected ServerSocketChannel javaChannel() {
   return (ServerSocketChannel) super.javaChannel();
}

super.javaChannel()

protected SelectableChannel javaChannel() {
    return ch;
}

之前已经看到了，创建的ServerSocketChannel被赋值给了ch，这一步就是获取原来的ch。

类型变化？继承关系如下

javaChannel也就是JDK原生的channel了。

javaChannel().socket()

就是原生ServerSocketChannel获取socket对象了，不是新东西。

NioServerSocketChannelConfig

private NioServerSocketChannelConfig(NioServerSocketChannel channel, ServerSocket javaSocket) {
     super(channel, javaSocket);
}

super

public DefaultServerSocketChannelConfig(ServerSocketChannel channel, ServerSocket javaSocket) {
   super(channel);
   if (javaSocket == null) {
       throw new NullPointerException("javaSocket");
   }
   this.javaSocket = javaSocket;
}

super之后把javaSocket带过来，旧东西不管，继续super

super

public DefaultChannelConfig(Channel channel) {
    this(channel, new AdaptiveRecvByteBufAllocator());
}

记住AdaptiveRecvByteBufAllocator，这是需要详细追查的。

为了流程完整性，我们继续深入。

DefaultChannelConfig

protected DefaultChannelConfig(Channel channel, RecvByteBufAllocator allocator) {
    setRecvByteBufAllocator(allocator, channel.metadata());
    this.channel = channel;
}

setRecvByteBufAllocator

private void setRecvByteBufAllocator(RecvByteBufAllocator allocator, ChannelMetadata metadata) {
    if (allocator instanceof MaxMessagesRecvByteBufAllocator) {
        ((MaxMessagesRecvByteBufAllocator) allocator).maxMessagesPerRead(metadata.defaultMaxMessagesPerRead());
    } else if (allocator == null) {
        throw new NullPointerException("allocator");
    }
    setRecvByteBufAllocator(allocator);
}

setRecvByteBufAllocator

public ChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
   rcvBufAllocator = checkNotNull(allocator, "allocator");
   return this;
}

也就是说，我们配置最关键的，就是DefaultChannelConfig新增的额外参数，allocator。

Allocator

前面已经说了，这个就是重点。

public DefaultChannelConfig(Channel channel) {
    this(channel, new AdaptiveRecvByteBufAllocator());
}

大小适配策略

public AdaptiveRecvByteBufAllocator() {
    this(DEFAULT_MINIMUM, DEFAULT_INITIAL, DEFAULT_MAXIMUM);
}

有参构造

    public AdaptiveRecvByteBufAllocator(int minimum, int initial, int maximum) {
        if (minimum <= 0) {
            throw new IllegalArgumentException("minimum: " + minimum);
        }
        if (initial < minimum) {
            throw new IllegalArgumentException("initial: " + initial);
        }
        if (maximum < initial) {
            throw new IllegalArgumentException("maximum: " + maximum);
        }
        int minIndex = getSizeTableIndex(minimum);
        if (SIZE_TABLE[minIndex] < minimum) {
            this.minIndex = minIndex + 1;
        } else {
            this.minIndex = minIndex;
        }
        int maxIndex = getSizeTableIndex(maximum);
        if (SIZE_TABLE[maxIndex] > maximum) {
            this.maxIndex = maxIndex - 1;
        } else {
            this.maxIndex = maxIndex;
        }
        this.initial = initial;
    }

静态代码块

static {
    List<Integer> sizeTable = new ArrayList<Integer>();
    for (int i = 16; i < 512; i += 16) {
        sizeTable.add(i);
    }
    for (int i = 512; i > 0; i <<= 1) {
        sizeTable.add(i);
    }
    SIZE_TABLE = new int[sizeTable.size()];
    for (int i = 0; i < SIZE_TABLE.length; i ++) {
        SIZE_TABLE[i] = sizeTable.get(i);
    }
}

可以分析出sizeTable的值为

1. $n\ast \left(2^4\right),n\in [16,512)$n∗(24),n∈[16,512)
2. $(2^8)^n, n \in[512, +\infty)$(28)n,n∈[512,+∞), $+\infty$+∞表示直到溢出.

getSizeTableIndex

    private static int getSizeTableIndex(final int size) {
        for (int low = 0, high = SIZE_TABLE.length - 1;;) {
            if (high < low) {
                return low;
            }
            if (high == low) {
                return high;
            }
            int mid = low + high >>> 1;
            int a = SIZE_TABLE[mid];
            int b = SIZE_TABLE[mid + 1];
            if (size > b) {
                low = mid + 1;
            } else if (size < a) {
                high = mid - 1;
            } else if (size == a) {
                return mid;
            } else {
                return mid + 1;
            }
        }
    }

此方法，其实通过size比对，找到一个接近并大于size的在SIZE_TABLE的值得坐标并返回。

有参构造调用，就是通过传入的限定字节大小，来确定SIZE_TABLE的index范围。

static final int DEFAULT_MINIMUM = 64;
static final int DEFAULT_INITIAL = 1024;
static final int DEFAULT_MAXIMUM = 65536;

public HandleImpl(int minIndex, int maxIndex, int initial) {
     this.minIndex = minIndex;
     this.maxIndex = maxIndex;
     index = getSizeTableIndex(initial);
     nextReceiveBufferSize = SIZE_TABLE[index];
}
public int guess() {
    return nextReceiveBufferSize;
}

public ByteBuf allocate(ByteBufAllocator alloc) {
     return alloc.ioBuffer(guess());
}

    public ByteBuf ioBuffer(int initialCapacity) {
        if (PlatformDependent.hasUnsafe()) {
            return directBuffer(initialCapacity);
        }
        return heapBuffer(initialCapacity);
    }

    public static boolean hasUnsafe() {
        return HAS_UNSAFE;
    }

private static final boolean HAS_UNSAFE = hasUnsafe0();

    private static boolean hasUnsafe0() {
        if (isAndroid()) {
            logger.debug("sun.misc.Unsafe: unavailable (Android)");
            return false;
        }
        if (PlatformDependent0.isExplicitNoUnsafe()) {
            return false;
        }
        try {
            boolean hasUnsafe = PlatformDependent0.hasUnsafe();
            logger.debug("sun.misc.Unsafe: {}", hasUnsafe ? "available" : "unavailable");
            return hasUnsafe;
        } catch (Throwable ignored) {
            return false;
        }
    }