0
点赞
收藏
分享

微信扫一扫

Netty——客户端Accpt源码分析

郝春妮 2023-03-21 阅读 95
nettyjava服务端非阻塞Html/CSS前端开发


摘要

本博文将介绍netty如何接受新的请求，netty如何给新请求分配reactor线程，netty如何给每个新连接增加ChannelHandler。

netty中的reactor线程

netty中最核心的东西莫过于两种类型的reactor线程，可以看作netty中两种类型的发动机，驱动着netty整个框架的运转。一种类型的reactor线程是boos线程组，专门用来接受新的连接，然后封装成channel对象扔给worker线程组；还有一种类型的reactor线程是worker线程组，专门用来处理连接的读写。

不管是boos线程还是worker线程，所做的事情均分为以下三个步骤

  1. 轮询注册在selector上的IO事件
  2. 处理IO事件
  3. 执行异步task

对于boos线程来说，第一步轮询出来的基本都是 accept 事件，表示有新的连接，而worker线程轮询出来的基本都是read/write事件，表示网络的读写事件。

新连接的建立

我们已经知道，当服务端绑启动之后，服务端的channel已经注册到boos reactor线程中，reactor不断检测有新的事件，直到检测出有accept事件发生。

NioEventLoop.java

private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final NioUnsafe unsafe = ch.unsafe();
    //检查该SelectionKey是否有效，如果无效，则关闭channel
    if (!k.isValid()) {
        // close the channel if the key is not valid anymore
        unsafe.close(unsafe.voidPromise());
        return;
    }

    try {
        int readyOps = k.readyOps();
        // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
        // to a spin loop
        // 如果准备好READ或ACCEPT则触发unsafe.read() ,检查是否为0，如上面的源码英文注释所说：解决JDK可能会产生死循环的一个bug。
        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
            if (!ch.isOpen()) {//如果已经关闭，则直接返回即可，不需要再处理该channel的其他事件
                // Connection already closed - no need to handle write.
                return;
            }
        }
        // 如果准备好了WRITE则将缓冲区中的数据发送出去，如果缓冲区中数据都发送完成，则清除之前关注的OP_WRITE标记
        if ((readyOps & SelectionKey.OP_WRITE) != 0) {
            // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
            ch.unsafe().forceFlush();
        }
        // 如果是OP_CONNECT，则需要移除OP_CONNECT否则Selector.select(timeout)将立即返回不会有任何阻塞，这样可能会出现cpu 100%
        if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
            // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
            // See https://github.com/netty/netty/issues/924
            int ops = k.interestOps();
            ops &= ~SelectionKey.OP_CONNECT;
            k.interestOps(ops);

            unsafe.finishConnect();
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

该方法主要是对SelectionKey k进行了检查，有如下几种不同的情况

  • 1）OP_ACCEPT，接受客户端连接
  • 2）OP_READ, 可读事件, 即 Channel 中收到了新数据可供上层读取。
  • 3）OP_WRITE, 可写事件, 即上层可以向 Channel 写入数据。
  • 4）OP_CONNECT, 连接建立事件, 即 TCP 连接已经建立, Channel 处于 active 状态。

看下当boss线程 selector检测到OP_ACCEPT事件时，内部干了些什么。

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
    unsafe.read();
    if (!ch.isOpen()) {
        //如果已经关闭，则直接返回即可，不需要再处理该channel的其他事件
        // Connection already closed - no need to handle write.
        return;
    }
}

boos reactor线程已经轮询到 ​​SelectionKey.OP_ACCEPT​​​ 事件，说明有新的连接进入，此时将调用channel的 ​​unsafe​​来进行实际的操作,此时的channel为 NioServerSocketChannel，则unsafe为NioServerSocketChannel的属性NioMessageUnsafe那么，我们进入到它的​​read​​方法，进入新连接处理的第二步。

注册到reactor线程

NioMessageUnsafe.java

private final List<Object> readBuf = new ArrayList<Object>();

public void read() {
    assert eventLoop().inEventLoop();
    final ChannelPipeline pipeline = pipeline();
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    do {
        //调用 doReadMessages 方法不断地读取消息，用 readBuf 作为容器
        int localRead = doReadMessages(readBuf);
        if (localRead == 0) {
            break;
        }
        if (localRead < 0) {
            closed = true;
            break;
        }
    } while (allocHandle.continueReading());
    int size = readBuf.size();
    for (int i = 0; i < size; i ++) {
        pipeline.fireChannelRead(readBuf.get(i));
    }
    readBuf.clear();
    pipeline.fireChannelReadComplete();
}

调用 ​​doReadMessages​​​ 方法不断地读取消息，用 ​​readBuf​​​ 作为容器，这里，其实可以猜到读取的是一个个连接，然后调用 ​​pipeline.fireChannelRead()​​​，将每条新连接经过一层服务端channel的洗礼，之后清理容器，触发 ​​pipeline.fireChannelReadComplete()​​

doReadMessages()

protected int doReadMessages(List<Object> buf) throws Exception {
    SocketChannel ch = javaChannel().accept();

    try {
        if (ch != null) {
            buf.add(new NioSocketChannel(this, ch));
            return 1;
        }
    } catch (Throwable t) {
        logger.warn("Failed to create a new channel from an accepted socket.", t);

        try {
            ch.close();
        } catch (Throwable t2) {
            logger.warn("Failed to close a socket.", t2);
        }
    }

    return 0;
}

我们终于窥探到netty调用jdk底层nio的边界 ​​javaChannel().accept();​​​，由于netty中reactor线程第一步就扫描到有accept事件发生，因此，这里的​​accept​​方法是立即返回的，返回jdk底层nio创建的一条channel。

ServerSocketChannel有阻塞和非阻塞两种模式：

  • a、阻塞模式：ServerSocketChannel.accept() 方法监听新进来的连接，当 accept()方法返回的时候,它返回一个包含新进来的连接的 SocketChannel。阻塞模式下, accept()方法会一直阻塞到有新连接到达。
  • b、非阻塞模式：，accept() 方法会立刻返回，如果还没有新进来的连接,返回的将是null。 因此，需要检查返回的SocketChannel是否是null.

在NioServerSocketChannel的构造函数分析中，我们知道，其通过ch.configureBlocking(false);语句设置当前的ServerSocketChannel为非阻塞的。

netty将jdk的 ​​SocketChannel​​​ 封装成自定义的 ​​NioSocketChannel​​，加入到list里面，这样外层就可以遍历该list，做后续处理，从上一篇文章中，我们已经知道服务端的创建过程中会创建netty中一系列的核心组件，包括pipeline,unsafe等等。

NioSocketChannel.java

public NioSocketChannel(Channel parent, SocketChannel socket) {
    super(parent, socket);
    config = new NioSocketChannelConfig(this, socket.socket());
}

我们重点分析 ​​super(parent, socket)，NioSocketChannel的父类为 AbstractNioByteChannel​​

AbstractNioByteChannel.java

protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
    super(parent, ch, SelectionKey.OP_READ);
}

这里，我们看到jdk nio里面熟悉的影子—— ​​SelectionKey.OP_READ​​，一般在原生的jdk nio编程中，也会注册这样一个事件，表示对channel的读感兴趣。

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的时候，最终也会进入到这个方法，​​super(parent)​​​, 便是在​​AbstractChannel​​中创建一系列和该channel绑定的组件，如下：

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

而这里的 ​​readInterestOp​​​ 表示该channel关心的事件是 ​​SelectionKey.OP_READ​​，后续会将该事件注册到selector，之后设置该通道为非阻塞模式，在channel中创建 unsafe 和一条 pipeline 。

pipeline.fireChannelRead(NioSocketChannel)

对于 ​​pipeline​​​​我们前面已经了解过，在netty的各种类型的channel中，都会包含一个pipeline，字面意思是管道，我们可以理解为一条流水线工艺，流水线工艺有起点，有结束，中间还有各种各样的流水线关卡，一件物品，在流水线起点开始处理，经过各个流水线关卡的加工，最终到流水线结束​​

对应到netty里面，流水线的开始就是​​HeadContxt​​​，流水线的结束就是​​TailConext​​​，​​HeadContxt​​​中调用​​Unsafe​​​做具体的操作，​​TailConext​​中用于向用户抛出pipeline中未处理异常以及对未处理消息的警告

通过前面的文章中，我们已经知道在服务端的channel初始化时，在pipeline中，已经自动添加了一个pipeline处理器 ​​ServerBootstrapAcceptor​​​, 并已经将用户代码中设置的一系列的参数传入了构造函数，接下来，我们就来看下​​ServerBootstrapAcceptor​​

ServerBootstrapAcceptor.java

private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {
    private final EventLoopGroup childGroup;
    private final ChannelHandler childHandler;
    private final Entry<ChannelOption<?>, Object>[] childOptions;
    private final Entry<AttributeKey<?>, Object>[] childAttrs;

    ServerBootstrapAcceptor(
            EventLoopGroup childGroup, ChannelHandler childHandler,
            Entry<ChannelOption<?>, Object>[] childOptions, Entry<AttributeKey<?>, Object>[] childAttrs) {
        this.childGroup = childGroup;
        this.childHandler = childHandler;
        this.childOptions = childOptions;
        this.childAttrs = childAttrs;
    }

    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        final Channel child = (Channel) msg;

        child.pipeline().addLast(childHandler);

        for (Entry<ChannelOption<?>, Object> e: childOptions) {
            try {
                if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {
                    logger.warn("Unknown channel option: " + e);
                }
            } catch (Throwable t) {
                logger.warn("Failed to set a channel option: " + child, t);
            }
        }

        for (Entry<AttributeKey<?>, Object> e: childAttrs) {
            child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
        }

        try {
            childGroup.register(child).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (!future.isSuccess()) {
                        forceClose(child, future.cause());
                    }
                }
            });
        } catch (Throwable t) {
            forceClose(child, t);
        }
    }
}

前面的 ​​pipeline.fireChannelRead(NioSocketChannel);​​​ 最终通过head->unsafe->ServerBootstrapAcceptor的调用链，调用到这里的 ​​ServerBootstrapAcceptor​​​ 的​​channelRead​​​方法，而 ​​channelRead​​​ 一上来就把这里的msg强制转换为 ​​Channel​​

然后，拿到该channel，也就是我们之前new出来的 ​​NioSocketChannel中​​​对应的pipeline，将用户代码中的 ​​childHandler​​​，添加到pipeline，这里的 ​​childHandler​​ 在用户代码中的体现为

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
 .channel(NioServerSocketChannel.class)
 .childHandler(new ChannelInitializer<SocketChannel>() {
     @Override
     public void initChannel(SocketChannel ch) throws Exception {
         ChannelPipeline p = ch.pipeline();
         p.addLast(new EchoServerHandler());
     }
 });

其实对应的是 ​​ChannelInitializer​​​，到了这里，​​NioSocketChannel​​中pipeline对应的处理器为 head->ChannelInitializer->tail，牢记，后面会再次提到！

接着，设置 ​​NioSocketChannel​​​ 对应的 attr和option，然后进入到 ​​childGroup.register(child)​​​，这里的childGroup就是我们在启动代码中new出来的​​NioEventLoopGroup​​

我们进入到​​NioEventLoopGroup​​​的​​register​​​方法，代理到其父类​​MultithreadEventLoopGroup​​

MultithreadEventLoopGroup.java

public ChannelFuture register(Channel channel) {
    return next().register(channel);
}



@Override
public EventLoop next() {
    return (EventLoop) super.next();
}

MultithreadEventExecutorGroup.java

public EventExecutor next() {
    return chooser.next();
}

这里的chooser对应的类为 ​​EventExecutorChooser​​，字面意思为事件执行器选择器，放到我们这里的上下文中的作用就是从worker reactor线程组中选择一个reactor线程。

public interface EventExecutorChooserFactory {

    /**
     * Returns a new {@link EventExecutorChooser}.
     */
    EventExecutorChooser newChooser(EventExecutor[] executors);

    /**
     * Chooses the next {@link EventExecutor} to use.
     */
    @UnstableApi
    interface EventExecutorChooser {

        /**
         * Returns the new {@link EventExecutor} to use.
         */
        EventExecutor next();
    }
}

chooser的实现有两种

public final class DefaultEventExecutorChooserFactory implements EventExecutorChooserFactory {

    public static final DefaultEventExecutorChooserFactory INSTANCE = new DefaultEventExecutorChooserFactory();

    private DefaultEventExecutorChooserFactory() { }

    @SuppressWarnings("unchecked")
    @Override
    public EventExecutorChooser newChooser(EventExecutor[] executors) {
        if (isPowerOfTwo(executors.length)) {
            return new PowerOfTowEventExecutorChooser(executors);
        } else {
            return new GenericEventExecutorChooser(executors);
        }
    }

    private static boolean isPowerOfTwo(int val) {
        return (val & -val) == val;
    }

    private static final class PowerOfTowEventExecutorChooser implements EventExecutorChooser {
        private final AtomicInteger idx = new AtomicInteger();
        private final EventExecutor[] executors;

        PowerOfTowEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
        }

        @Override
        public EventExecutor next() {
            return executors[idx.getAndIncrement() & executors.length - 1];
        }
    }

    private static final class GenericEventExecutorChooser implements EventExecutorChooser {
        private final AtomicInteger idx = new AtomicInteger();
        private final EventExecutor[] executors;

        GenericEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
        }

        @Override
        public EventExecutor next() {
            return executors[Math.abs(idx.getAndIncrement() % executors.length)];
        }
    }
}

默认情况下，chooser通过 ​​DefaultEventExecutorChooserFactory​​​被创建，在创建reactor线程选择器的时候，会判断reactor线程的个数，如果是2的幂，就创建​​PowerOfTowEventExecutorChooser​​​，否则，创建​​GenericEventExecutorChooser​​

两种类型的选择器在选择reactor线程的时候，都是通过Round-Robin的方式选择reactor线程，唯一不同的是，​​PowerOfTowEventExecutorChooser​​​是通过与运算，而​​GenericEventExecutorChooser​​​是通过取余运算，与运算的效率要高于求余运算，选择完一个reactor线程，即 ​​NioEventLoop​​ 之后，我们回到注册的地方。

public ChannelFuture register(Channel channel) {
    return next().register(channel);
}

SingleThreadEventLoop.java

#其实，这里已经和服务端启动的过程一样了

@Override
public ChannelFuture register(Channel channel) {
    return register(new DefaultChannelPromise(channel, this));
}

private void register0(ChannelPromise promise) {
    boolean firstRegistration = neverRegistered;
    doRegister();
    neverRegistered = false;
    registered = true;

    pipeline.invokeHandlerAddedIfNeeded();

    safeSetSuccess(promise);
    pipeline.fireChannelRegistered();
    if (isActive()) {
        if (firstRegistration) {
            pipeline.fireChannelActive();
        } else if (config().isAutoRead()) {
            beginRead();
        }
    }
}

protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            selectionKey = javaChannel().register(eventLoop().selector, 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                eventLoop().selectNow();
                selected = true;
            } else {
                throw e;
            }
        }
    }
}

将该条channel绑定到一个​​selector​​​上去，一个selector被一个reactor线程使用，后续该channel的事件轮询，以及事件处理，异步task执行都是由此reactor线程来负责，绑定完reactor线程之后，调用 ​​pipeline.invokeHandlerAddedIfNeeded()。​​​前面我们说到，到目前为止​​NioSocketChannel​​​ 的pipeline中有三个处理器，head->ChannelInitializer->tail，最终会调用到 ​​ChannelInitializer​​​ 的 ​​handlerAdded​​ 方法。

public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    if (ctx.channel().isRegistered()) {
        initChannel(ctx);
    }
}

​​handlerAdded​​​方法调用 ​​initChannel​​​ 方法之后，调用​​remove(ctx);​​将自身删除，如下

AbstractNioChannel.java

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
    if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { 
        try {
            initChannel((C) ctx.channel());
        } catch (Throwable cause) {
            exceptionCaught(ctx, cause);
        } finally {
            remove(ctx);
        }
        return true;
    }
    return false;
}

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
 .channel(NioServerSocketChannel.class)
 .option(ChannelOption.SO_BACKLOG, 100)
 .handler(new LoggingHandler(LogLevel.INFO))
 .childHandler(new ChannelInitializer<SocketChannel>() {
     @Override
     public void initChannel(SocketChannel ch) throws Exception {
         ChannelPipeline p = ch.pipeline();
         p.addLast(new LoggingHandler(LogLevel.INFO));
         p.addLast(new EchoServerHandler());
     }
 });

注册读事件

private void register0(ChannelPromise promise) {
    // ..
    pipeline.fireChannelRegistered();
    if (isActive()) {
        if (firstRegistration) {
            pipeline.fireChannelActive();
        } else if (config().isAutoRead()) {
            beginRead();
        }
    }
}

​​pipeline.fireChannelRegistered();​​​，其实没有干啥有意义的事情，最终无非是再调用一下业务pipeline中每个处理器的 ​​ChannelHandlerAdded​​​方法处理下回调，​​isActive()​​​在连接已经建立的情况下返回true，所以进入方法块，进入到 ​​pipeline.fireChannelActive();​​在这里我详细步骤先省略，直接进入到关键环节。

AbstractNioChannel.java

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

    readPending = true;

    final int interestOps = selectionKey.interestOps();
    if ((interestOps & readInterestOp) == 0) {
        selectionKey.interestOps(interestOps | readInterestOp);
    }
}

这里其实就是将 ​​SelectionKey.OP_READ​​事件注册到selector中去，表示这条通道已经可以开始处理read事件了。

客户端Accpt分析总结

  • 1.boos reactor线程轮询到有新的连接进入
  • 2.通过封装jdk底层的channel创建 ​​NioSocketChannel​​以及一系列的netty核心组件
  • 3.将该条连接通过chooser，选择一条worker reactor线程绑定上去
  • 4.注册读事件，开始新连接的读写
举报

相关推荐

Netty源码分析 - Bootstrap客户端

netty个人学习一些收藏Netty

Netty WebSocket客户端

.netbootstrapideLinux系统/运维

python netty客户端

客户端PythonHTTP后端开发

nsq客户端源码分析

日记本golang学习篇章

Java Netty websocket 客户端

java

Redis 3.0源码分析-客户端

Redis客户端服务器redis虚拟化云计算

netty参考 android手机客户端

.netxmlandroid系统/运维

客户端 UDP 调用netty 参考

.netideipad虚拟化云计算

基于netty封装socket客户端

python开发语言后端.netios虚拟化云计算

客户端 TCP 调用netty 参考

.netidebootstrap系统/运维
0 条评论