netty 源码分析（三） Channel

作者: sschrodinger | 来源:发表于2019-06-27 09:38 被阅读0次

netty 源码分析（三） Channel

sschrodinger

2019/06/25

参考

《Netty 权威指南》第二版 - 李林锋著

JDK version 1.8

Channel 簇

Channel

Channel 是 netty 抽象出来的与网络 I/O ，该接口实现了所有的 I/O 操作方法和与 netty 框架相关的方法，如 pipeline() 等方法获得责任链。Channel 的定义如下：

public interface ChannelOutboundInvoker {

    ChannelFuture bind(SocketAddress localAddress);

    ChannelFuture connect(SocketAddress remoteAddress);

    ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress);

    ChannelFuture disconnect();

    ChannelFuture close();

    ChannelFuture deregister();

    ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise);

    ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise);

    ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);

    ChannelFuture disconnect(ChannelPromise promise);

    ChannelFuture close(ChannelPromise promise);

    ChannelFuture deregister(ChannelPromise promise);

    ChannelOutboundInvoker read();

    ChannelFuture write(Object msg);

    ChannelFuture write(Object msg, ChannelPromise promise);

    ChannelOutboundInvoker flush();

    ChannelFuture writeAndFlush(Object msg, ChannelPromise promise);

    ChannelFuture writeAndFlush(Object msg);

    ChannelPromise newPromise();

    ChannelProgressivePromise newProgressivePromise();

    ChannelFuture newSucceededFuture();

    ChannelFuture newFailedFuture(Throwable cause);

    ChannelPromise voidPromise();
}

public interface Channel extends AttributeMap, ChannelOutboundInvoker, Comparable<Channel> {

    ChannelId id();

    EventLoop eventLoop();

    Channel parent();

    ChannelConfig config();

    boolean isOpen();

    boolean isRegistered();

    boolean isActive();

    ChannelMetadata metadata();

    SocketAddress localAddress();

    SocketAddress remoteAddress();

    ChannelFuture closeFuture();

    boolean isWritable();

    long bytesBeforeUnwritable();

    long bytesBeforeWritable();

    Unsafe unsafe();

    ChannelPipeline pipeline();

    ByteBufAllocator alloc();

    @Override
    Channel read();

    @Override
    Channel flush();

    interface Unsafe {
        //...
    }
}

note

ChannelOutboundInvoker 接口申明了所有的外向的网络操作，即流量向外的网络操作。

Channel 接口返回的都是 Future 类型的变量及其子类（立即返回），通过 Future 的 get() 方法或者 isDone() 方法判断是否执行成功。（包括 ChannelFuture、ChannelPromise、ChannelProgressivePromise 等以 future 或者 promise 结尾的都是 Future 的子类）。

对于所有的 I/O 操作，都是异步的，所以需要返回 Future 对返回的结果进行处理。

AbstractChannel 实现了部分的 Channel 接口，AbstractChannel 抽象类包括了如下域：

// 一些静态的异常
private static final ClosedChannelException ENSURE_OPEN_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new ExtendedClosedChannelException(null), AbstractUnsafe.class, "ensureOpen(...)");
private static final ClosedChannelException CLOSE_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new ClosedChannelException(), AbstractUnsafe.class, "close(...)");
private static final ClosedChannelException WRITE_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new ExtendedClosedChannelException(null), AbstractUnsafe.class, "write(...)");
private static final ClosedChannelException FLUSH0_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new ExtendedClosedChannelException(null), AbstractUnsafe.class, "flush0()");
private static final NotYetConnectedException FLUSH0_NOT_YET_CONNECTED_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new NotYetConnectedException(), AbstractUnsafe.class, "flush0()");

// 父 channel
private final Channel parent;
private final ChannelId id;
private final Unsafe unsafe;
// 默认的 pipeline
private final DefaultChannelPipeline pipeline;
// 空的 future，对于 isDone, isSuccess 等方法，都立即返回 false
private final VoidChannelPromise unsafeVoidPromise = new VoidChannelPromise(this, false);
private final CloseFuture closeFuture = new CloseFuture(this);

private volatile SocketAddress localAddress;
private volatile SocketAddress remoteAddress;
private volatile EventLoop eventLoop;
private volatile boolean registered;
private boolean closeInitiated;
private Throwable initialCloseCause;

/** Cache for the string representation of this channel */
private boolean strValActive;
    private String strVal;

每一个域都可以实现特定的功能，通过组合，就可以实现 Channel 的各种功能。

基本上继承自 ChannelOutboundInvoker 的方法，都交由 pipeline 处理，如下：

public ChannelFuture bind(SocketAddress localAddress) {
    return pipeline.bind(localAddress);

    
//...
}

同时，该类也增加了一些较为实用的函数供子类改写：


protected abstract void doBeginRead() throws Exception;
// 该方法会在 Channel 被注册到 EventLoop（一个线程）中时被调用
protected void doRegister() throws Exception {
    // NOOP
}

protected abstract void doBind(SocketAddress localAddress) throws Exception;

protected abstract void doDisconnect() throws Exception;
protected abstract void doClose() throws Exception;

protected void doDeregister() throws Exception {
    // NOOP
}

AbstractNioChannel

该类主要实现了 JDK 中 NIO 的部分功能。

首先来看该类所持有的部分域：

// NIO 可选择 Channel
private final SelectableChannel ch;
protected final int readInterestOp;
// NIO 感兴趣的键
volatile SelectionKey selectionKey;
boolean readPending;

/**
 * The future of the current connection attempt.  If not null, subsequent
 * connection attempts will fail.
 */
private ChannelPromise connectPromise;
// 返回一个 ScheduleFuture，相比普通的 Future，多了一个 getDelay 方法 
private ScheduledFuture<?> connectTimeoutFuture;
private SocketAddress requestedRemoteAddress;

首先看该类的构造函数，如下：

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    super(parent);
    this.ch = ch;
    this.readInterestOp = readInterestOp;
    try {
        ch.configureBlocking(false);
    } catch (IOException e) {
        try {
            ch.close();
        } catch (IOException e2) {
            if (logger.isWarnEnabled()) {
                logger.warn(
                        "Failed to close a partially initialized socket.", e2);
            }
        }

        throw new ChannelException("Failed to enter non-blocking mode.", e);
    }
}

最关键的一点就是在初始化该类的实例时，会将 channel 设置成非阻塞模式的。

重点看 doRegister 方法，该方法主要是开启监听，并将该 Channel 注册到 EventLoop （一个线程，用于处理事件）。

// return SelectableChannel
protected SelectableChannel javaChannel() {return ch;}

@Override
protected void doRegister() throws Exception {
    boolean selected = false;
    // 3.
    for (;;) {
        try {
            // 1.
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                // Force the Selector to select now as the "canceled" SelectionKey may still be
                // cached and not removed because no Select.select(..) operation was called yet.
                // 2.
                eventLoop().selectNow();
                selected = true;
            } else {
                // We forced a select operation on the selector before but the SelectionKey is still cached
                // for whatever reason. JDK bug ?
                throw e;
            }
        }
    }
}

第一处 selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this) 实际上是调用的 channel.register(selector, 0, object)。

这就是 NIO 中将 channel 注册到 selector 的步骤，接下来的两个参数，0 代表对任意事件都不关心，object 代表 attachment （附件），在该方法中，将他自身作为附件，在以后可以通过 SelectionKey 类的 attachment 方法得到该类自身。

如下是一个 attachment 的样例代码：

public class Demo {

    public static void main(String[] args) throws IOException {
        Selector selector = Selector.open();
        SocketChannel channel = SocketChannel.open();
        channel.configureBlocking(false);
        SelectionKey key = channel.register(selector, 0, new Attachment());
        System.out.println(((Attachment)key.attachment()).value);

    }

    public static class Attachment {
        int value = 10;
    }
}

// output
// 10

如果在某次操作中，将 SelectKey 取消，即调用了 key.cancel() 函数，就会导致在注册时抛出 CancelledKeyException 异常（），如下：

public class Demo {

    public static void main(String[] args) throws IOException {
        Selector selector = Selector.open();
        SocketChannel channel = SocketChannel.open();
        channel.configureBlocking(false);
        System.out.println("=========== first register =============");
        SelectionKey key = channel.register(selector, 0, new Attachment());
        System.out.println("=========== cancel key =================");
        key.cancel();
        System.out.println("=========== second register ============");
        channel.register(selector, 0, new Attachment());

    }
}
// output
// =========== first register =============
// =========== cancel key =================
// =========== second register ============
// Exception in thread "main" java.nio.channels.CancelledKeyException

原因是因为调用 key.cancel() 之后，会将 channel 置为无效，需要刷新。可以使用 selectNow 刷新，将原来的 channel 删除掉。如下：

Selector selector = Selector.open();
SocketChannel channel = SocketChannel.open();
channel.configureBlocking(false);
System.out.println("=========== first register =============");
SelectionKey key = channel.register(selector, 0, new Attachment());
System.out.println("=========== cancel key =================");
key.cancel();
System.out.println("=========== select now =================");
selector.selectNow();
System.out.println("=========== second register ============");
channel.register(selector, 0, new Attachment());

如果取消掉也仍然抛出异常，则是 JDK 的 bug，直接抛出。

第三步使用了 selected 变量和一个 for 循环控制注册，基本等效于如上的示例代码。

doDeregister() 方法主要就是调用了 key.cancel()，略。

doBeginRead() 方法主要实现了切换监听的兴趣为读，表示开始监听读事件。如下：

@Override
protected void doBeginRead() throws Exception {
    // Channel.read() or ChannelHandlerContext.read() was called
    // 判断 selectionKey 是否有效
    final SelectionKey selectionKey = this.selectionKey;
    if (!selectionKey.isValid()) {
        return;
    }

    readPending = true;

    final int interestOps = selectionKey.interestOps();
    // 增加读的兴趣字
    if ((interestOps & readInterestOp) == 0) {
        selectionKey.interestOps(interestOps | readInterestOp);
    }
}

对于 doClose() 方法，实现如下：

protected void doClose() throws Exception {
    ChannelPromise promise = connectPromise;
    if (promise != null) {
        // Use tryFailure() instead of setFailure() to avoid the race against cancel().
        promise.tryFailure(DO_CLOSE_CLOSED_CHANNEL_EXCEPTION);
        connectPromise = null;
    }

    ScheduledFuture<?> future = connectTimeoutFuture;
    if (future != null) {
        future.cancel(false);
        connectTimeoutFuture = null;
    }
}

connectPromise 代表的是正在进行连接的操作，如果不为空，则证明连接还没有成功，所以设定关闭已经关闭的通道异常，最后调用 connectTimeoutFuture 的 cancel 函数关闭。

该类添加了两个新方法供子类改写：

// 连接到服务器的操作
protected abstract boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception;

// 连接完成之后执行的操作
protected abstract void doFinishConnect() throws Exception;

AbstractNioByteChannel

AbstractNioByteChannel 实现了 AbstractChannel 的 doWrite 方法，该方法主要的目的就是将在缓冲区内的数据发送到远端。

对于一个普通的 NIO 程序，发送数据的方法示例代码如下：

public class Demo {

    public static void main(String[] args) throws IOException {
        SocketChannel channel = SocketChannel.open();
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        byteBuffer.put(DATA);
        byteBuffer.flip();
        channel.write(byteBuffer);
    }

    public static final byte DATA = '0';

}

以上代码会将 '0' 作为数据发送至远程端。

对于 netty 来说，他的实现代码如下：

@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
    int writeSpinCount = config().getWriteSpinCount();
    do {
        Object msg = in.current();
        if (msg == null) {
            // Wrote all messages.
            clearOpWrite();
            // Directly return here so incompleteWrite(...) is not called.
            return;
        }
        writeSpinCount -= doWriteInternal(in, msg);
        } while (writeSpinCount > 0);

    incompleteWrite(writeSpinCount < 0);
}

ChannelOutboundBuffer 代表的是发送缓冲区，可以理解为多个 ByteBuf 的一个数组。

writeSpinCount 代表的是循环发送次数，可以理解为一次发送 ByteBuf 数组的多少个元素。

note

设置 writeSpinCount 的主要目的是当循环发送时， I/O 线程会一直执行写操作， I/O 线程就不能执行其他的操作，如果网络阻塞，可能导致线程假死。

每次循环从 ChannelOutboundBuffer 中读取一个对象，如果对象为空，说明所有的对象都已经发送成功了，这个时候调用 clearOpWrite() 将 selectionKey 的写兴趣字去掉，并直接返回。clearOpWrite 如下，主要就是取消写的兴趣字：

protected final void clearOpWrite() {
    final SelectionKey key = selectionKey();
    // Check first if the key is still valid as it may be canceled as part of the deregistration
    // from the EventLoop
    // See https://github.com/netty/netty/issues/2104
    if (!key.isValid()) {
        return;
    }
    final int interestOps = key.interestOps();
    if ((interestOps & SelectionKey.OP_WRITE) != 0) {
        key.interestOps(interestOps & ~SelectionKey.OP_WRITE);
    }
}

如果消息不为空且循环次数仍然大于 0，则调用 doWriteInternal() 方法发送缓冲区的数据，实现如下：

private int doWriteInternal(ChannelOutboundBuffer in, Object msg) throws Exception {
    if (msg instanceof ByteBuf) {
        ByteBuf buf = (ByteBuf) msg;
        if (!buf.isReadable()) {
            in.remove();
            return 0;
        }

        final int localFlushedAmount = doWriteBytes(buf);
        if (localFlushedAmount > 0) {
            in.progress(localFlushedAmount);
            if (!buf.isReadable()) {
                in.remove();
            }
            return 1;
        }
    } else if (msg instanceof FileRegion) {
        //...
    } else {
            // Should not reach here.
            throw new Error();
    }
    // WRITE_STATUS_SNDBUF_FULL = 2147483647
    return WRITE_STATUS_SNDBUF_FULL;
}

如果 msg 是 ByteBuf 类型：

判断 buf 是否可读（是否有数据），如果没有，删除他执行下一次循环
利用 doWriteBytes(buf) 函数发送数据，返回实际上发送的字节数
如果发送的字节数实际上大于 0，则发送一个通知（in.progress(localFlushedAmount)）
如果发送完成之后，buf 不可读，则说明已经发送完，进入下一循环
如果发送的字节数等于 0，则可能是由于没有发送的缓存窗口了，直接返回一个大数

note

in.progress() 实际上是调用了一个特殊的 Future，该 Future 会在收到消息后就给监听者发送通知，发送的通知为发送了多少字节的数据，有点类似于进度条的感觉。

监听者可以根据通知（进度条信息）做一些特定的事情。

回到 doWrite 函数，如果没有发送完所有数据就达到了循环发送的最高次数，则会调用 incompleteWrite(writeSpinCount < 0) 函数，设置 channel 的 OP_WRITE 兴趣字、将发送挂在任务中等操作。代码如下：

protected final void incompleteWrite(boolean setOpWrite) {
    // Did not write completely.
    if (setOpWrite) {
        setOpWrite();
    } else {
        // It is possible that we have set the write OP, woken up by NIO because the socket is writable, and then
        // use our write quantum. In this case we no longer want to set the write OP because the socket is still
        // writable (as far as we know). We will find out next time we attempt to write if the socket is writable
        // and set the write OP if necessary.
        clearOpWrite();

        // Schedule flush again later so other tasks can be picked up in the meantime
        eventLoop().execute(flushTask);
    }
}

经过分析，参数 setOpWirite 只在 doWriteInternal 函数返回 WRITE_STATUS_SNDBUF_FULL 时才会为 true，说明出现了 send buff full，不可写，则设置写监听的套接字（setOpWrite()），直到可写。其他情况下，可以继续发送，但是为了给其他任务执行的机会，就将其加入到线程池的 execute 队列中，待稍后执行。

note

eventLoop().execute(flushTask) 可以等效为 executor.execute(() -> doWrite(in))，其中 executor 就是一个普通的线程池。即一段时间后继续执行发送函数。

除了 doWriteBytes()需要子类实现，该类也增加其他函数供子类实现：

protected abstract int doReadBytes(ByteBuf buf) throws Exception;

NioSocketChannel

该类是客户端 Channel 的最终实现类。

主要实现了 doReadBytes、doConnect、doBind、doFinishConnect、doDisconnect、doClose、doWriteBytes 等方法。

doBind 方法实现如下：

@Override
protected void doBind(SocketAddress localAddress) throws Exception {
    doBind0(localAddress);
}
private void doBind0(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        SocketUtils.bind(javaChannel(), localAddress);
    } else {
        SocketUtils.bind(javaChannel().socket(), localAddress);
    }
}
public static void bind(final SocketChannel socketChannel, final SocketAddress address) throws IOException {
    try {
        AccessController.doPrivileged(new PrivilegedExceptionAction<Void>() {
            @Override
            public Void run() throws IOException {
                socketChannel.bind(address);
                return null;
            }
        });
    } catch (PrivilegedActionException e) {
        throw (IOException) e.getCause();
    }
}

最终是调用的 JDK 的 channel.bind(InetAddress) 方法，略。

doConnect 的实现如下：

@Override
protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
    if (localAddress != null) {
        doBind0(localAddress);
    }

    boolean success = false;
    try {
        // 等效于 channel.connect(remoteAddress)
        boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
        if (!connected) {
            selectionKey().interestOps(SelectionKey.OP_CONNECT);
        }
        success = true;
        return connected;
    } finally {
        if (!success) {
            doClose();
        }
    }
}

对于一个非阻塞的 channel（AbstractChannel 在初始化时就将 cannel 设置成了非阻塞），如果调用 channel.connect() 方法，其返回值有三种情况：

如果连接立即建立，则返回 true
如果连接暂时没有建立（服务端没有返回 ACK 应答等，连接结果不确定），则返回 false
如果连接失败，直接抛出 I/O 异常

对于连接操作，第一步是绑定端口，即第一个 if 语句。

接下来连接至远端，即 SocketUtils.connect(javaChannel(), remoteAddress)。

如果返回 true，表明连接成功，返回成功。

如果返回 false，表明还未成功，在该 channel 中加入连接完成的兴趣字。

如果抛出了异常，会执行 finally 语句的 doClose 方法，关闭连接。

note

如果对兴趣字增加了连接完成的监听

如果 selector 返回，表示连接通道连接就绪或者发生了错误。要使用finishConnect判断下，如果连接失败，会抛出异常

一般来说，NIO 处理就绪 OP_CONNECT 的通用代码如下：

if (key.isValid() && key.isConnectable()) {
    SocketChannel ch = (SocketChannel) key.channel();
    if (ch.finishConnect()) {
        // Connect successfully
        // key.interestOps(SelectionKey.OP_READ);
} else {
    // Connect failed
}

doFinishConnect、doDiconnect、doClose、doWriteBytes、doReadBytes 如下：

// 与 channel 原生的 finishConnect 不同，这直接抛出异常
@Override
protected void doFinishConnect() throws Exception {
    if (!javaChannel().finishConnect()) {
        throw new Error();
    }
}

@Override
protected void doDisconnect() throws Exception {
    doClose();
}

@Override
protected void doClose() throws Exception {
    super.doClose();
    javaChannel().close();
}

@Override
protected int doWriteBytes(ByteBuf buf) throws Exception {
    final int expectedWrittenBytes = buf.readableBytes();
    // 直接向 channel 写数据
    return buf.readBytes(javaChannel(), expectedWrittenBytes);
}

@Override
protected int doReadBytes(ByteBuf byteBuf) throws Exception {
    // 设置一个足够大的缓冲区读数据
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.attemptedBytesRead(byteBuf.writableBytes());
    // 读数据，并将数据保存在 byteBuf 中
    return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
}

对于服务器端的 channel 来说，还有 AbstractNioMessageChannel 和 NioServerSocketChannel 两个类，暂时跳过。

==综上，对于 Channel 簇来说，除了由 Channel 接口提供的函数，交由 pipeline 实现之外，还提供了各种 doXXX 命名的帮助函数。==

Unsafe 簇

Unsafe 类是 Channel 的内部类，只允许非用户线程访问。该接口的定义如下：

interface Unsafe {
    // 主要在读数据时生成 ByteBuf
    RecvByteBufAllocator.Handle recvBufAllocHandle();

    SocketAddress localAddress();

    SocketAddress remoteAddress();

    // 在某个 eventLoop （线程）上注册该 channel 的 promise，一旦完成，通知 ChannelFuture
    void register(EventLoop eventLoop, ChannelPromise promise);

    void bind(SocketAddress localAddress, ChannelPromise promise);

    void connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);

    void disconnect(ChannelPromise promise);

    void close(ChannelPromise promise);

    void closeForcibly();

    void deregister(ChannelPromise promise);

    // 计划一个读取操作，该操作将填充Channel管道中第一个ChannelInbindingHandler的入站缓冲区。如果已有挂起的读取操作，则此方法不执行任何操作。
    void beginRead();

    void write(Object msg, ChannelPromise promise);

    void flush();

    // 一个占位的 promise
    ChannelPromise voidPromise();

    // 返回发送缓冲
    ChannelOutboundBuffer outboundBuffer();
}

AbstractUnsafe

AbstractUnsafe 是所有 Unsafe 类的抽象类，他定义了如下实用方法：

// 是否被注册并且在当前 eventLoop 中
private void assertEventLoop() {
    assert !registered || eventLoop.inEventLoop();
}

先看 register 方法，如下：

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    if (eventLoop == null) {
        throw new NullPointerException("eventLoop");
    }
    if (isRegistered()) {
        promise.setFailure(new IllegalStateException("registered to an event loop already"));
        return;
    }
    if (!isCompatible(eventLoop)) {
        promise.setFailure(
                new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
        return;
    }

    AbstractChannel.this.eventLoop = eventLoop;

    if (eventLoop.inEventLoop()) {
        register0(promise);
    } else {
        try {
            eventLoop.execute(new Runnable() {
                @Override
                public void run() {
                    register0(promise);
                }
            });
        } catch (Throwable t) {
            logger.warn(
                    "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                    AbstractChannel.this, t);
            closeForcibly();
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }
}

首先对当前状态进行一些判断

如果已经注册，抛出异常
如果不兼容，抛出异常

最后，判断是否是当前线程在执行 eventLoop

是的话，就在当前线程注册
否则，将注册封装到一个 task 中，交由线程池运行

以下是注册实现的主要逻辑，如下：

private void register0(ChannelPromise promise) {
    try {
        // check if the channel is still open as it could be closed in the mean time when the register
        // call was outside of the eventLoop
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }
        boolean firstRegistration = neverRegistered;
        doRegister();
        neverRegistered = false;
        registered = true;

        // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
        // user may already fire events through the pipeline in the ChannelFutureListener.
        pipeline.invokeHandlerAddedIfNeeded();

        safeSetSuccess(promise);
        pipeline.fireChannelRegistered();
        // Only fire a channelActive if the channel has never been registered. This prevents firing
        // multiple channel actives if the channel is deregistered and re-registered.
        if (isActive()) {
            if (firstRegistration) {
                pipeline.fireChannelActive();
            } else if (config().isAutoRead()) {
                // This channel was registered before and autoRead() is set. This means we need to begin read
                // again so that we process inbound data.
                //
                // See https://github.com/netty/netty/issues/4805
                beginRead();
            }
        }
    } catch (Throwable t) {
        // Close the channel directly to avoid FD leak.
        closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}

该函数的主要逻辑是调用 Channel 提供的 doRegister 方法，将 Channel 注册到 selector 中。

如果 doRegister 没有抛出异常，则说明注册到了 selector 上，则将 ChannelPromise 的状态设置为 success。

设置成功后，向 pipeline 发送 fireChannelRegistered() 事件，供上层处理。

然后，判断当前 Channel 的状态，即 isActive 方法。该方法会在 Channel 是打开并且已经连接的情况下返回 true。

在激活状态下，如果是第一次注册，即向 pipline 发送一个 fireChannelActive 事件。

如果不是第一次注册，说明之前有注册过，如果是自动读取的配置，则会进行读取（beginRead() 方法）。

beginRead 方法主要是调用 doBeginRead 方法设置读的感兴趣字。

write 方法主要是用于向发送缓冲中增加元素。

@Override
public final void write(Object msg, ChannelPromise promise) {
    // ...
    outboundBuffer.addMessage(msg, size, promise);
}

flush 方法主要是将发送缓冲的数据发送出去，代码如下：

public final void flush() {
    // ...
    // 标识之前的数据都需要发送
    outboundBuffer.addFlush();
    // 发送核心逻辑
    flush0();
}
protected void flush0() {
    if (inFlush0) {
        // Avoid re-entrance
        return;
    }
    //...
    inFlush0 = true;

    // Mark all pending write requests as failure if the channel is inactive.
    if (!isActive()) {
        // ...
    }

    try {
        // 调用 Channel 的 doWrite 发送数据
        doWrite(outboundBuffer);
    } catch (Throwable t) {
        // ...
    } finally {
        inFlush0 = false;
    }
}

AbstractNioUnsafe

AbstractNioUnsafe 主要实现了 connect 方法，如下：

@Override
public final void connect(
        final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
    // ...

    try {
        // 1. 
        if (connectPromise != null) {
            // Already a connect in process.
            throw new ConnectionPendingException();
        }

        boolean wasActive = isActive();
        // 2.
        if (doConnect(remoteAddress, localAddress)) {
            // 3.
            fulfillConnectPromise(promise, wasActive);
        } else {
            // 4. 
            connectPromise = promise;
            requestedRemoteAddress = remoteAddress;

            // Schedule connect timeout.
            int connectTimeoutMillis = config().getConnectTimeoutMillis();
            if (connectTimeoutMillis > 0) {
                connectTimeoutFuture = eventLoop().schedule(new Runnable() {
                    @Override
                    public void run() {
                        ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                        ConnectTimeoutException cause =
                                new ConnectTimeoutException("connection timed out: " + remoteAddress);
                        if (connectPromise != null && connectPromise.tryFailure(cause)) {
                            close(voidPromise());
                        }
                    }
                }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
            }

            promise.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isCancelled()) {
                        if (connectTimeoutFuture != null) {
                            connectTimeoutFuture.cancel(false);
                        }
                        connectPromise = null;
                        close(voidPromise());
                    }
                }
            });
        }
    } catch (Throwable t) {
        promise.tryFailure(annotateConnectException(t, remoteAddress));
        closeIfClosed();
    }    
}

主要分成了 4 个步骤。

判断是否已经存在正在连接的操作，如果存在，抛出异常
调用 doConnect 进行连接，如果成功，返回 true，如果没有立即获得链接，返回 false
如果立即获得了链接，调用 fullfilConnectionPromise 方法。该方法的主要目的就是将 promise 的状态设置为 sueccess，同时，如果是第一次激活，发出 fireChannelActive 事件
如果没有立即连接成功，会将等待的任务加到事件监听中

对于步骤四
首先构建了一个超时处理的任务，如果到 time out 的时间，这个任务都没有被取消，说明超超时了，就会执行超时之后的逻辑。如下：

Runnable runnable = new Runnable() {
    @Override
    public void run() {
        ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
        ConnectTimeoutException cause = new ConnectTimeoutException("connection timed out: " + remoteAddress);
        if (connectPromise != null && connectPromise.tryFailure(cause)) {
            close(voidPromise());
        }
    }
}

connectTimeoutFuture = eventLoop().schedule(runnable, , connectTimeoutMillis, TimeUnit.MILLISECONDS);

那么如果在超时之前连接成功，就需要取消该任务。netty 的做法是添加一个事件监听，当监听到连接成功时，取消该任务，如下：

promise.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        if (future.isCancelled()) {
            if (connectTimeoutFuture != null) {
                connectTimeoutFuture.cancel(false);
            }
            connectPromise = null;
            close(voidPromise());
        }
    });

finishConnect 代码如下：

@Override
public final void finishConnect() {
    // Note this method is invoked by the event loop only if the connection attempt was
    // neither cancelled nor timed out.

    assert eventLoop().inEventLoop();

    try {
        boolean wasActive = isActive();
        doFinishConnect();
        fulfillConnectPromise(connectPromise, wasActive);
    } catch (Throwable t) {
        fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
    } finally {
        // Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
        // See https://github.com/netty/netty/issues/1770
        if (connectTimeoutFuture != null) {
            connectTimeoutFuture.cancel(false);
        }
        connectPromise = null;
    }
}

AbstractNioByteUnsafe

AbstractNioByteUnsafe 主要实现了 read 方法。实现如下：

@Override
public final void read() {
    // ...

    ByteBuf byteBuf = null;
    boolean close = false;
    try {
        do {
            // 分配一个 ByteBuf 用于接收数据
            byteBuf = allocHandle.allocate(allocator);
            // 接受数据
            allocHandle.lastBytesRead(doReadBytes(byteBuf));
            // 获得最近添加的字节数
            if (allocHandle.lastBytesRead() <= 0) {
                // 如果字节数小于等于 0，代表什么都没有收到，释放缓冲区
                byteBuf.release();
                byteBuf = null;
                close = allocHandle.lastBytesRead() < 0;
                if (close) {
                    // 如果字节数小于 0，说明已经没有可读数据，将readingPending（尝试读）设为 false
                            readPending = false;
                }
                // 返回
                break;
            }

            // 如果有读到数据，则对读到的数据数进行计数 + 1
            allocHandle.incMessagesRead(1);
            readPending = false;
            // 触发可读事件
            pipeline.fireChannelRead(byteBuf);
            byteBuf = null;
        } while (allocHandle.continueReading());

        // 读取完成后触发读完成事件
        allocHandle.readComplete();
        pipeline.fireChannelReadComplete();

        if (close) {
            closeOnRead(pipeline);
        }
    } catch (Throwable t) {
        handleReadException(pipeline, byteBuf, t, close, allocHandle);
    } finally {
        // Check if there is a readPending which was not processed yet.
        // This could be for two reasons:
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
        //
        // See https://github.com/netty/netty/issues/2254
        if (!readPending && !config.isAutoRead()) {
            removeReadOp();
        }
    }
}

==综上，Unsafe 封装了一系列对 I/O 的操作方法，可以通过这些方法，操作 I/O。==

如：某线程可以的调用调用链可能如下：

public void demo() {
    ChannelPromise promise;
    EventLoop eventLoop;
    // 注册到 selector
    unsafe.register(eventLoop, promise);
    // 连接到远程地址
    unsafe.connect(remoteAddress, localAddress, promise);
    // 发送数据
    unsfafe.write();
    // 接受数据
    unsafe.read();
}

总结

Channel 接口和 Unsafe 接口实现了用于 I/O 操作的大部分方法，方便 netty 其他线程对 I/O 进行读写，大大降低了工作难度。Channel 操作的特点如下：

所有的 Channel 接口的操作都是异步的：返回一个 Future 对象用于后期处理或者直接返回空，在函数内部设置 Future 对象
对于 NIO 的空轮询 bug，使用重建的方式对 Selector 进行重建
可打断的发送方式，在发送一定时间后会主动退出，给其他操作运行的机会（doWrite() 方法）

netty 源码分析（三） Channel

netty 源码分析（三） Channel

参考

Channel 簇

Channel

AbstractNioChannel

AbstractNioByteChannel

NioSocketChannel

Unsafe 簇

AbstractUnsafe

AbstractNioUnsafe

AbstractNioByteUnsafe

总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

netty 源码分析 （三） Channel

netty 源码分析 （三） Channel

参考

Channel 簇

Channel

AbstractNioChannel

AbstractNioByteChannel

NioSocketChannel

Unsafe 簇

AbstractUnsafe

AbstractNioUnsafe

AbstractNioByteUnsafe

总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

netty 源码分析（三） Channel

netty 源码分析（三） Channel