Netty源码分析第3章(客户端接入流程)—->第4节: NioSocketChannel注册到selector

Netty源码分析第三章: 客户端接入流程

第四节: NioSocketChannel注册到selector

我们回到最初的NioMessageUnsafe的read()方法:

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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
1public void read() {

2    //必须是NioEventLoop方法调用的, 不能通过外部线程调用

3    assert eventLoop().inEventLoop();

4    //服务端channel的config

5    final ChannelConfig config = config();

6    //服务端channel的pipeline

7    final ChannelPipeline pipeline = pipeline();

8    //处理服务端接入的速率

9    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();

10    //设置配置

11    allocHandle.reset(config);

12    boolean closed = false;

13    Throwable exception = null;

14    try {

15        try {

16            do {

17                //创建jdk底层的channel

18                //readBuf用于临时承载读到链接

19                int localRead = doReadMessages(readBuf);

20                if (localRead == 0) {

21                    break;

22                }

23                if (localRead < 0) {

24                    closed = true;

25                    break;

26                }

27                //分配器将读到的链接进行计数

28                allocHandle.incMessagesRead(localRead);

29                //连接数是否超过最大值

30            } while (allocHandle.continueReading());

31        } catch (Throwable t) {

32            exception = t;

33        }

34        int size = readBuf.size();

35        //遍历每一条客户端连接

36        for (int i = 0; i < size; i ++) {

37            readPending = false;

38            //传递事件, 将创建NioSokectChannel进行传递

39            //最终会调用ServerBootstrap的内部类ServerBootstrapAcceptor的channelRead()方法

40            pipeline.fireChannelRead(readBuf.get(i));

41        }

42        readBuf.clear();

43        allocHandle.readComplete();

44        pipeline.fireChannelReadComplete();

45        //代码省略

46    } finally {

47        //代码省略

48    }

49}

50

在while循环结束之后, 将会通过一个for循环遍历readBuf集合, 并将创建的NioSocketChannel传入fireChannelRead()中, 传播channel的读取事件

有关pipeline的知识, 我们下一章会详细剖析, 并会根据剖析后的内容回顾之前的有关pipeline的操作, 这里我们只需知道, 通过fireChannelRead()我们最终调用了ServerBootstrap的内部类ServerBootstrapAcceptor 中的channelRead()方法

跟到channelRead()方法中:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
1public void channelRead(ChannelHandlerContext ctx, Object msg) {

2    final Channel child = (Channel) msg;

3    //代码省略

4    try {

5        //work线程注册channel

6        childGroup.register(child).addListener(new ChannelFutureListener() {

7            @Override

8            public void operationComplete(ChannelFuture future) throws Exception {

9                if (!future.isSuccess()) {

10                    forceClose(child, future.cause());

11                }

12            }

13        });

14    } catch (Throwable t) {

15        forceClose(child, t);

16    }

17}

18

其中参数的msg就是最初传入fireChannelRead()方法的NioSocketChannel

所以这里可以通过 
final Channel child = (Channel) msg 这种方式拿到NioSocketChannel

其中childGroup是我们最初初始化的work线程, 这里的register()方法跟boss线程一样, 通过next()方法获选择一个线程进行注册, 这里不再赘述

我们紧跟调用链, 跟到SingleThreadEventLoop的register()方法:

1
2
3
4
5
6
1public ChannelFuture register(final ChannelPromise promise) {

2    ObjectUtil.checkNotNull(promise, "promise");

3    promise.channel().unsafe().register(this, promise);

4    return promise;

5}

6

这里的unsafe(), 根据我们之前的剖析, 是NioByteUnsafe, 这里的register最终会调用AbstractUnsafe的register()方法, 并NioSocketChannel

不知道同学们是否记得, 当初NioServerSocketChannel注册的时候也走的这个方法

我们跟到register()这个方法中:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
1public final void register(EventLoop eventLoop, final ChannelPromise promise) {

2    //省略验证代码

3    //所有的复制操作, 都交给eventLoop处理

4    AbstractChannel.this.eventLoop = eventLoop;

5

6    if (eventLoop.inEventLoop()) {

7        //做实际主注册

8        register0(promise);

9    } else {

10        try {

11            eventLoop.execute(new Runnable() {

12                @Override

13                public void run() {

14                    register0(promise);

15                }

16            });

17        } catch (Throwable t) {

18            //代码省略

19        }

20    }

21}

22

我们学习过NioEventLoop相关知识之后, 应该对这部分代码不太陌生, 首先判断是不是当前NioEventLoop线程, 如果是, 则直接进行注册操作, 如果不是, 则封装成task在当前NioEventLoop中执行

走到这里不难明白, 这里并不是当前NioEventLoop线程, 这是boss线程执行的, 所以这里会走到else, 如果是第一次的连接操作, work线程的NioEventLoop并没有启动, 所以这里也会启动NioEventLoop, 并开始轮询操作

跟到register0(promise)中看其是如何做实际操作的:

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
1private void register0(ChannelPromise promise) {

2    try {

3        //省略代码

4        //做实际的注册

5        doRegister();

6        neverRegistered = false;

7        registered = true;

8        //触发事件

9        pipeline.invokeHandlerAddedIfNeeded();

10        safeSetSuccess(promise);

11        //触发注册成功事件

12        pipeline.fireChannelRegistered();

13        if (isActive()) {

14            if (firstRegistration) {

15                //传播active事件(4)

16                pipeline.fireChannelActive();

17            } else if (config().isAutoRead()) {

18                beginRead();

19            }

20        }

21    } catch (Throwable t) {

22        //省略代码

23    }

24}

25

这段代码我们同样并不陌生, 因为NioServerSokectChannel中也走这一部分, 我们继续关注doRegister()方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
1protected void doRegister() throws Exception {

2    boolean selected = false;

3    for (;;) {

4        try {

5            //jdk底层的注册方法

6            //第一个参数为selector, 第二个参数表示不关心任何事件

7            selectionKey = javaChannel().register(eventLoop().selector, 0, this);

8            return;

9        } catch (CancelledKeyException e) {

10            //省略代码

11        }

12    }

13}

14

这部分也是我们之前剖析过的jdk底层的注册, 只是不同的是, 这里的javaChannel()是SocketChanel而不是ServerSocketChannel

同样, 这里也是表示不关心任何事件, 只是在当前NioEventLoop绑定的selector上注册

至此, NioSocketChannel完成注册

上一节: NioSocketChannel的创建

下一节: 监听读事件