从零开始实现简易版Netty(一) MyNetty Reactor模式

从零开始实现简易版Netty(一) MyNetty Reactor模式

自从18年作为一个java程序员入行以来，所接触到的大量组件如dubbo、rocketmq、redisson等都是基于netty这一高性能网络框架实现的。
限于个人水平，在过去很长一段时间中都只能算是netty的初级使用者；在使用基于netty的中间件时，总是因为对netty底层不够了解而导致排查问题时效率不高。
因此，在过去的一段时间中我对netty源码进行了一定的研究，并以博客的形式将心得分享出来，希望能帮助到同样对netty工作原理感兴趣的读者。


非常感谢大佬bin的技术小屋，在我学习netty的过程中给了我很大的帮助。
1. MyNetty介绍

不同于大多数博客直接针对netty官方源码进行解析的方式，本系列博客通过从零到一的实现一个简易版的netty(即MyNetty)来帮助读者更好的理解netty的工作原理。
相比于完整版的netty，MyNetty只实现了netty中最核心的功能点，目的是降低复杂度，避免初学者在学习netty的过程中，对netty源码中复杂的抽象及过深的调用链感到畏惧。
本博客会按照以下顺序，通过一个接一个的小迭代由简单到复杂的实现MyNetty，每一个迭代都会有一篇与之对应的技术博客。

• Reactor模式
  
• Pipeline管道
  
• 高效的数据读取
  
• 高效的数据写出
  
• FastThreadLocal
  
• ByteBuf
  
• Normal级别的池化内存分配(伙伴算法)
  
• Small级别的池化内存分配(slab算法)
  
• 池化内存分配支持线程本地缓存(ThreadLocalCache)
  
• 常用的编解码器(FixedLengthFrameDecoder/LineBasedFrameDecoder等)
  

MyNetty的核心逻辑主要参考自netty 4.1.80.Final版本。
2. 操作系统I/O模型与Reactor模式介绍

作为MyNetty系列的第一篇博客，按照规划，第一个迭代中需要实现基于NIO的reactor模式。这也是netty最核心的功能，一个基于事件循环的reactor线程工作模型。


在学习的过程中，我们要尽量做到知其然且知其所以然。
因此，在介绍Reactor模式之前，先简单介绍一下两种常见的操作系统网络I/O模型，只要在了解其各自的优缺点后，才能帮助我们更好的理解为什么Netty最终选择了reactor模式。
2.1 操作系统I/O模型介绍

同步阻塞I/O(BIO)

同步阻塞IO，顾名思义，其读写是阻塞性的，在数据还没有准备好时(比如客户端还未发送新请求,或者未收到服务端响应)，当前处理IO的线程是处于阻塞态的，直到数据就绪(比如接受到客户端发送的请求，或收到服务端响应)时才会被唤醒。
由于其阻塞的特性，因此在服务端并发时，每一个新的客户端连接都需要一个独立的线程来承载。


BIO详情优点简单易理解，同步阻塞式的线性代码执行流符合人的直觉。因此普通的web业务后台服务器大多是基于BIO模型开发的缺点由于客户端连接数与服务器线程数是1:1的，而服务器由于线程上下文切换的CPU开销和内存大小限制，难以应对大规模的并发连接(大几千甚至几万)，性能较差BIO服务端demo

1. public class BIOEchoServer {
3.     private static final ExecutorService threadPool = Executors.newCachedThreadPool();
5.     public static void main(String[] args) throws IOException {
6.         int port = 8080;
7.         ServerSocket serverSocket = new ServerSocket(port);
8.         System.out.println("BIOServer started on port " + port);
10.         while (true) {
11.             Socket clientSocket = serverSocket.accept();
12.             System.out.println("New client connected: " + clientSocket.getInetAddress());
14.             // 每个新的连接都启用一个线程去处理
15.             threadPool.execute(
16.                 () -> handleClientConnect(clientSocket)
17.             );
18.         }
19.     }
21.     private static void handleClientConnect(Socket clientSocket) {
22.         try (BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
23.              PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true)) {
25.             String inputLine;
26.             while ((inputLine = in.readLine()) != null) {
27.                 System.out.println("Received from client: " + inputLine);
29.                 // echo message
30.                 String responseMessage = "server echo: " + inputLine;
31.                 out.println(responseMessage);
32.                 System.out.println("Sent response: " + responseMessage);
33.             }
34.         } catch (IOException e) {
35.             System.out.println("Client connection closed: " + e.getMessage());
36.         } finally {
37.             try {
38.                 clientSocket.close();
39.                 System.out.println("clientSocket closed! " + clientSocket.getInetAddress());
40.             } catch (IOException e) {
41.                 System.err.println("Error closing client socket: " + e.getMessage());
42.             }
43.         }
44.     }
45. }

复制代码

BIO客户端demo

1. public class BIOClient {
3.     public static void main(String[] args) throws IOException {
4.         String hostname = "127.0.0.1";
5.         int port = 8080;
7.         try (Socket socket = new Socket(hostname, port);
8.              PrintWriter out = new PrintWriter(socket.getOutputStream(), true);
9.              BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
10.              BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in))) {
12.             System.out.println("Connected to server. Type messages (type 'exit' to quit)");
14.             String userInput;
15.             while ((userInput = stdIn.readLine()) != null) {
16.                 out.println(userInput);
17.                 System.out.println("Server response: " + in.readLine());
18.             }
19.         }
20.     }
21. }

复制代码

I/O多路复用

I/O多路复用，顾名思义，其不同于BIO中一个线程对应一个客户端连接的模式。I/O多路复用模型中，一个服务端线程能够同时处理多个客户端连接。
I/O多路复用解决了传统BIO模型下面对海量并发时系统资源不足的问题，但同时也引入了一些新的问题。


I/O多路复用详情优点性能好，吞吐量高。单个线程即可处理海量连接缺点比起BIO的阻塞模式，基于事件触发的编程模型非常复杂。IO多路复用服务端demo

1. public class NIOEchoServer {
3.     public static void main(String[] args) throws IOException {
4.         SelectorProvider selectorProvider = SelectorProvider.provider();
5.         Selector selector = selectorProvider.openSelector();
7.         // 服务端监听accept事件的channel
8.         ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
9.         serverSocketChannel.socket().bind(new InetSocketAddress(8080));
10.         serverSocketChannel.configureBlocking(false);
11.         serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
13.         for(;;){
14.             try{
15.                 int keys = selector.select(60000);
16.                 if (keys == 0) {
17.                     System.out.println("server 60s未监听到事件，继续监听！");
18.                     continue;
19.                 }
21.                 // processSelectedKeysPlain
22.                 Iterator<SelectionKey> selectionKeyItr = selector.selectedKeys().iterator();
23.                 while (selectionKeyItr.hasNext()) {
24.                     SelectionKey key = selectionKeyItr.next();
25.                     System.out.println("process SelectionKey=" + key.readyOps());
26.                     try {
28.                         // 拿出来后，要把集合中已经获取到的事件移除掉，避免重复的处理
29.                         selectionKeyItr.remove();
31.                         if (key.isAcceptable()) {
32.                             // 处理accept事件（接受到来自客户端的连接请求）
33.                             processAcceptEvent(key);
34.                         }
36.                         if (key.isReadable()) {
37.                             // 处理read事件
38.                             processReadEvent(key);
39.                         }
40.                     }catch (Exception e){
41.                         System.out.println("server event loop process an selectionKey error! " + e.getMessage());
42.                         e.printStackTrace();
44.                         key.cancel();
45.                         if(key.channel() != null){
46.                             System.out.println("has error, close channel! " + key.channel());
47.                             key.channel().close();
48.                         }
49.                     }
50.                 }
51.             }catch (Exception e){
52.                 System.out.println("server event loop error! ");
53.                 e.getStackTrace();
54.             }
55.         }
56.     }
58.     private static void processAcceptEvent(SelectionKey key) throws IOException {
59.         // 能收到accept事件的channel一定是ServerSocketChannel
61.         ServerSocketChannel ssChannel = (ServerSocketChannel)key.channel();
62.         // 获得与客户端建立的那个连接
63.         SocketChannel socketChannel = ssChannel.accept();
64.         socketChannel.configureBlocking(false);
66.         socketChannel.finishConnect();
68.         System.out.println("socketChannel=" + socketChannel + " finishConnect!");
69.         // 将接受到的连接注册到同样的selector中，并监听read事件
70.         socketChannel.register(key.selector(),SelectionKey.OP_READ);
71.     }
73.     private static void processReadEvent(SelectionKey key) throws IOException {
74.         SocketChannel socketChannel = (SocketChannel)key.channel();
76.         // 简单起见，buffer不缓存，每次读事件来都新创建一个
77.         // 暂时也不考虑黏包/拆包场景(Netty中靠ByteToMessageDecoder解决，后续再分析其原理)，理想的认为每个消息都小于1024，且每次读事件都只有一个消息
78.         ByteBuffer readBuffer = ByteBuffer.allocate(1024);
80.         int byteRead = socketChannel.read(readBuffer);
81.         if(byteRead == -1){
82.             // 简单起见不考虑tcp半连接的情况，返回-1直接关掉连接
83.             socketChannel.close();
84.         }else{
85.             // 将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作
86.             readBuffer.flip();
87.             // 根据缓冲区可读字节数创建字节数组
88.             byte[] bytes = new byte[readBuffer.remaining()];
89.             // 将缓冲区可读字节数组复制到新建的数组中
90.             readBuffer.get(bytes);
91.             String receivedStr = new String(bytes, StandardCharsets.UTF_8);
93.             System.out.println("received message:" + receivedStr + " ,from " + socketChannel.socket().getRemoteSocketAddress());
95.             // 读完了，echo服务器准备回写数据到客户端
96.             String echoMessage = "server echo:" + receivedStr;
98.             ByteBuffer writeBuffer = ByteBuffer.allocateDirect(1024);
99.             writeBuffer.put(echoMessage.getBytes(StandardCharsets.UTF_8));
100.             writeBuffer.flip(); // 写完了，flip供后续去读取
101.             socketChannel.write(writeBuffer);
102.         }
103.     }
104. }

复制代码

IO多路复用客户端demo

1. public class NIOClient {
3.     private static volatile SocketChannel clientSocketChannel;
5.     public static void main(String[] args) throws Exception {
6.         SelectorProvider selectorProvider = SelectorProvider.provider();
7.         Selector selector = selectorProvider.openSelector();
9.         CountDownLatch countDownLatch = new CountDownLatch(1);
10.         new Thread(()->{
11.             try {
12.                 startClient(selector,countDownLatch);
13.             } catch (IOException e) {
14.                 e.printStackTrace();
15.             }
16.         }).start();
18.         countDownLatch.await();
19.         System.out.println("please input message:");
20.         while(true){
21.             Scanner sc = new Scanner(System.in);
22.             String msg = sc.next();
23.             System.out.println("get input message:" + msg);
25.             // 发送消息
26.             ByteBuffer writeBuffer = ByteBuffer.allocate(64);
27.             writeBuffer.put(msg.getBytes(StandardCharsets.UTF_8));
28.             writeBuffer.flip(); // 写完了，flip供后续去读取
29.             clientSocketChannel.write(writeBuffer);
30.         }
31.     }
33.     private static void startClient(Selector selector, CountDownLatch countDownLatch) throws IOException {
34.         SocketChannel socketChannel = SocketChannel.open();
35.         socketChannel.configureBlocking(false);
37.         clientSocketChannel = socketChannel;
39.         // doConnect
40.         // Returns: true if a connection was established,
41.         //          false if this channel is in non-blocking mode and the connection operation is in progress;
42.         if(!socketChannel.connect(new InetSocketAddress("127.0.0.1", 8080))) {
43.             // 配置为非阻塞，会返回false，通过注册并监听connect事件的方式进行交互
44.             socketChannel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
45.         }
47.         for(;;){
48.             try {
49.                 int keys = selector.select(60000);
50.                 if (keys == 0) {
51.                     System.out.println("client 60s未监听到事件，继续监听！");
52.                     continue;
53.                 }
55.                 // processSelectedKeysPlain
56.                 Iterator<SelectionKey> selectionKeyItr = selector.selectedKeys().iterator();
58.                 while (selectionKeyItr.hasNext()) {
59.                     SelectionKey key = selectionKeyItr.next();
60.                     try {
61.                         System.out.println("process SelectionKey=" + key.readyOps());
63.                         // 拿出来后，要把集合中已经获取到的事件移除掉，避免重复的处理
64.                         selectionKeyItr.remove();
66.                         if (key.isConnectable()) {
67.                             // 处理连接相关事件
68.                             processConnectEvent(key,countDownLatch);
69.                         }
71.                         if (key.isReadable()){
72.                             processReadEvent(key);
73.                         }
75.                         if (key.isWritable()){
76.                             System.out.println("watch an write event!");
77.                         }
79.                     } catch (Exception e) {
80.                         System.out.println("client event loop process an selectionKey error! " + e.getMessage());
82.                         key.cancel();
83.                         if(key.channel() != null){
84.                             key.channel().close();
85.                             System.out.println("has error, close channel!" );
86.                         }
87.                     }
88.                 }
89.             } catch (Exception e) {
90.                 System.out.println("client event loop error! ");
91.                 e.getStackTrace();
92.             }
93.         }
94.     }
96.     private static void processConnectEvent(SelectionKey key, CountDownLatch countDownLatch) throws IOException {
97.         // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
98.         int ops = key.interestOps();
99.         ops &= ~SelectionKey.OP_CONNECT;
100.         key.interestOps(ops);
102.         SocketChannel socketChannel = (SocketChannel) key.channel();
103.         if(socketChannel.finishConnect()){
104.             // 确认完成连接
105.             System.out.println("client channel connected!");
107.             countDownLatch.countDown();
108.         }else{
109.             // 连接建立失败，程序退出
110.             System.out.println("client channel connect failed!");
111.             System.exit(1);
112.         }
113.     }
115.     private static void processReadEvent(SelectionKey key) throws IOException {
116.         SocketChannel socketChannel = (SocketChannel) key.channel();
118.         // 创建ByteBuffer，并开辟一个1M的缓冲区
119.         ByteBuffer buffer = ByteBuffer.allocate(64);
120.         // 读取请求码流，返回读取到的字节数
121.         int readBytes = socketChannel.read(buffer);
123.         // 读取到字节，对字节进行编解码
124.         if(readBytes > 0){
125.             // 将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作
126.             buffer.flip();
127.             // 根据缓冲区可读字节数创建字节数组
128.             byte[] bytes = new byte[buffer.remaining()];
129.             // 将缓冲区可读字节数组复制到新建的数组中
130.             buffer.get(bytes);
131.             String response = new String(bytes, StandardCharsets.UTF_8);
132.             System.out.println("client received response message: " + response);
133.         }
135.         // 读取到了EOF，关闭连接
136.         if(readBytes < 0){
137.             socketChannel.close();
138.         }
139.     }
140. }

复制代码

上述对于操作系统I/O模型的介绍限于篇幅，点到为止。想进一步了解的读者可以参考我之前写的博客：谈谈对不同I/O模型的理解
2.2 Reactor模式

从上面的介绍中我们可以看到，I/O多路复用模型的高性能、高吞吐的特点更加适合互联网时代海量连接的场景，所以netty自然也是基于I/O多路复用模型的。
但上述给出的I/O多路复用的demo中存在两个很严重的问题，第一个问题是java中NIO的能力过于底层，在开发业务时所需要考虑的细节太多,一个简单的、不考虑各种异常、边界场景的echo服务器都要写近百行的代码。
第二个问题则是服务端单线程的I/O多路复用模型没法很好的利用现代的多核CPU硬件，会出现处理大量连接时一核有难八核围观的问题。


针对第一个问题，正是netty作为java NIO的更高层次封装而诞生的原因，我们会在后续的迭代中逐步的优化这一问题。
而第二个问题的解决方案便是本章要引出的主题，reactor模式。


I/O多路复用模型与多线程并不冲突，一个线程可以独自处理所有连接，也可以用多个线程来均匀的分摊所有来自客户端的连接。
在reactor模式下，接收连接与处理连接后续读写的任务的线程会被分离开。接受客户端连接的逻辑较为简单，因此一个线程(cpu核心)通常足够处理这一任务。
相对的，处理连接建立后的读写操作则压力会大的多，所以需要多个CPU核心(多个线程)来分摊压力。
在reactor模式下，将专门用于接受连接的线程称为Boss线程，而连接建立后处理读写操作的线程成为Worker线程(Boss工作压力小，Worker工作压力大；Boss接了单子后把活直接派给Worker)。
reactor模式示意图

3. MyNetty reactor模式实现源码解析

从上文IO多路复用的demo可以看到，程序最核心的逻辑便是处理selector.select获取到的事件key集合。
当前线程会不断地尝试获取到激活的事件集合，然后按顺序处理，并循环往复。这一工作机制被称为事件循环(EventLoop)。
事件被抽象为4种类型，OP_READ(可读事件)、OP_WRITE(可写事件)、OP_CONNECT(连接建立事件)和OP_ACCEPT(连接接受事件)，而在demo中我们已经接触到了除了OP_WRITE事件外的三种(OP_WRITE事件会在lab4高效的数据写出中再展开介绍)。
针对事件循环，Netty中抽象出了两个概念，EventLoopGroup和EventLoop，EventLoop对应的就是上述的无限循环处理IO事件的线程，而EventLoopGroup顾名思义便是将一组EventLoop统一管理的集合。


下面我们结合MyNetty的源码，来进一步讲解reactor模式的工作原理。
MyNetty NioServer源码

1. public class MyNettyNioServer {
3.     private static final Logger logger = LoggerFactory.getLogger(MyNettyNioServer.class);
5.     private final InetSocketAddress endpointAddress;
7.     private final MyNioEventLoopGroup bossGroup;
9.     public MyNettyNioServer(InetSocketAddress endpointAddress, MyEventHandler myEventHandler,
10.                             int bossThreads, int childThreads) {
11.         this.endpointAddress = endpointAddress;
13.         MyNioEventLoopGroup childGroup = new MyNioEventLoopGroup(myEventHandler,childThreads);
14.         this.bossGroup = new MyNioEventLoopGroup(myEventHandler,bossThreads,childGroup);
15.     }
17.     public void start() throws IOException {
18.         ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
19.         serverSocketChannel.configureBlocking(false);
21.         MyNioEventLoop myNioEventLoop = this.bossGroup.next();
23.         myNioEventLoop.execute(()->{
24.             try {
25.                 Selector selector = myNioEventLoop.getUnwrappedSelector();
26.                 serverSocketChannel.socket().bind(endpointAddress);
27.                 SelectionKey selectionKey = serverSocketChannel.register(selector, 0);
28.                 // 监听accept事件
29.                 selectionKey.interestOps(selectionKey.interestOps() | SelectionKey.OP_ACCEPT);
30.                 logger.info("MyNioServer do start! endpointAddress={}",endpointAddress);
31.             } catch (IOException e) {
32.                 logger.error("MyNioServer do bind error!",e);
33.             }
34.         });
35.     }
36. }

复制代码

MyNetty NioClient源码

1. public class MyNettyNioClient {
3.     private static final Logger logger = LoggerFactory.getLogger(MyNettyNioClient.class);
5.     private final InetSocketAddress remoteAddress;
7.     private final MyNioEventLoopGroup eventLoopGroup;
9.     private SocketChannel socketChannel;
11.     public MyNettyNioClient(InetSocketAddress remoteAddress, MyEventHandler myEventHandler, int nThreads) {
12.         this.remoteAddress = remoteAddress;
14.         this.eventLoopGroup = new MyNioEventLoopGroup(myEventHandler,nThreads);
15.     }
17.     public void start() throws IOException {
18.         SocketChannel socketChannel = SocketChannel.open();
19.         socketChannel.configureBlocking(false);
21.         this.socketChannel = socketChannel;
23.         MyNioEventLoop myNioEventLoop = this.eventLoopGroup.next();
25.         myNioEventLoop.execute(()->{
26.             try {
27.                 Selector selector = myNioEventLoop.getUnwrappedSelector();
29.                 // doConnect
30.                 // Returns: true if a connection was established,
31.                 //          false if this channel is in non-blocking mode and the connection operation is in progress;
32.                 if(!socketChannel.connect(remoteAddress)){
33.                     SelectionKey selectionKey = socketChannel.register(selector, 0);
34.                     int clientInterestOps = SelectionKey.OP_CONNECT | SelectionKey.OP_READ;
35.                     selectionKey.interestOps(selectionKey.interestOps() | clientInterestOps);
36.                 }
38.                 // 监听connect事件
39.                 logger.info("MyNioClient do start! remoteAddress={}",remoteAddress);
40.             } catch (IOException e) {
41.                 logger.error("MyNioClient do connect error!",e);
42.             }
43.         });
44.     }
45. }

复制代码

MyNetty EventLoop源码

1. public class MyNioEventLoop implements Executor {
3.     private static final Logger logger = LoggerFactory.getLogger(MyNioEventLoop.class);
5.     /**
6.      * 原始的jdk中的selector
7.      * */
8.     private final Selector unwrappedSelector;
10.     private final Queue<Runnable> taskQueue = new LinkedBlockingQueue<>(16);
12.     private volatile Thread thread;
13.     private final MyNioEventLoopGroup childGroup;
15.     private final AtomicBoolean threadStartedFlag = new AtomicBoolean(false);
17.     private MyEventHandler myEventHandler;
19.     public MyNioEventLoop(){
20.         this(null);
21.     }
23.     public MyNioEventLoop(MyNioEventLoopGroup childGroup) {
24.         this.childGroup = childGroup;
26.         SelectorProvider selectorProvider = SelectorProvider.provider();
27.         try {
28.             this.unwrappedSelector = selectorProvider.openSelector();
29.         } catch (IOException e) {
30.             throw new RuntimeException("open selector error!",e);
31.         }
32.     }
34.     @Override
35.     public void execute(Runnable task) {
36.         // 将任务加入eventLoop所属的任务队列，事件循环中会
37.         taskQueue.add(task);
39.         if(this.thread != Thread.currentThread()){
40.             // 如果执行execute方法的线程不是当前线程，可能当前eventLoop对应的thread还没有启动
41.             // 尝试启动当前eventLoop对应的线程(cas防并发)
42.             if(threadStartedFlag.compareAndSet(false,true)){
43.                 // 类似netty的ThreadPerTaskExecutor,启动一个线程来执行事件循环
44.                 new Thread(()->{
45.                     // 将eventLoop的thread与新启动的这个thread进行绑定
46.                     this.thread = Thread.currentThread();
48.                     // 执行监听selector的事件循环
49.                     doEventLoop();
50.                 }).start();
51.             }
52.         }
53.     }
55.     public Selector getUnwrappedSelector() {
56.         return unwrappedSelector;
57.     }
59.     public void setMyEventHandler(MyEventHandler myEventHandler) {
60.         this.myEventHandler = myEventHandler;
61.     }
63.     private void doEventLoop(){
64.         // 事件循环
65.         for(;;){
66.             try{
67.                 if(taskQueue.isEmpty()){
68.                     int keys = unwrappedSelector.select(60000);
69.                     if (keys == 0) {
70.                         logger.info("server 60s未监听到事件，继续监听！");
71.                         continue;
72.                     }
73.                 }else{
74.                     // 确保任务队列里的任务能够被触发
75.                     unwrappedSelector.selectNow();
76.                 }
78.                 // 简单起见，暂不实现基于时间等元素的更为公平的执行策略
79.                 // 直接先处理io，再处理所有task(ioRatio=100)
80.                 try {
81.                     // 处理监听到的io事件
82.                     processSelectedKeys();
83.                 }finally {
84.                     // Ensure we always run tasks.
85.                     // 处理task队列里的任务
86.                     runAllTasks();
87.                 }
88.             }catch (Throwable e){
89.                 logger.error("server event loop error!",e);
90.             }
91.         }
92.     }
94.     private void processSelectedKeys() throws IOException {
95.         // processSelectedKeysPlain
96.         Iterator<SelectionKey> selectionKeyItr = unwrappedSelector.selectedKeys().iterator();
97.         while (selectionKeyItr.hasNext()) {
98.             SelectionKey key = selectionKeyItr.next();
99.             logger.info("process SelectionKey={}",key.readyOps());
100.             try {
101.                 // 拿出来后，要把集合中已经获取到的事件移除掉，避免重复的处理
102.                 selectionKeyItr.remove();
104.                 if (key.isConnectable()) {
105.                     // 处理客户端连接建立相关事件
106.                     processConnectEvent(key);
107.                 }
109.                 if (key.isAcceptable()) {
110.                     // 处理服务端accept事件（接受到来自客户端的连接请求）
111.                     processAcceptEvent(key);
112.                 }
114.                 if (key.isReadable()) {
115.                     // 处理read事件
116.                     processReadEvent(key);
117.                 }
118.             }catch (Throwable e){
119.                 logger.error("server event loop process an selectionKey error!",e);
121.                 // 处理io事件有异常，取消掉监听的key，并且尝试把channel也关闭掉
122.                 key.cancel();
123.                 if(key.channel() != null){
124.                     logger.error("has error, close channel={} ",key.channel());
125.                     key.channel().close();
126.                 }
127.             }
128.         }
129.     }
131.     private void runAllTasks(){
132.         for (;;) {
133.             // 通过无限循环，直到把队列里的任务全部捞出来执行掉
134.             Runnable task = taskQueue.poll();
135.             if (task == null) {
136.                 return;
137.             }
139.             try {
140.                 task.run();
141.             } catch (Throwable t) {
142.                 logger.warn("A task raised an exception. Task: {}", task, t);
143.             }
144.         }
145.     }
147.     private void processAcceptEvent(SelectionKey key) throws IOException {
148.         ServerSocketChannel ssChannel = (ServerSocketChannel)key.channel();
150.         SocketChannel socketChannel = ssChannel.accept();
151.         if(this.childGroup != null){
152.             // boss/worker模式，boss线程只负责接受和建立连接
153.             // 将建立的连接交给child线程组去处理后续的读写
154.             MyNioEventLoop childEventLoop = childGroup.next();
155.             childEventLoop.execute(()->{
156.                 doRegister(childEventLoop,socketChannel);
157.             });
158.         }else{
159.             doRegister(this,socketChannel);
160.         }
161.     }
163.     private void processConnectEvent(SelectionKey key) throws IOException {
164.         // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
165.         // See https://github.com/netty/netty/issues/924
166.         int ops = key.interestOps();
167.         ops &= ~SelectionKey.OP_CONNECT;
168.         key.interestOps(ops);
170.         SocketChannel socketChannel = (SocketChannel) key.channel();
171.         if(socketChannel.finishConnect()){
172.             // 确认完成连接
173.             logger.info("client channel connected! socketChannel={}",socketChannel);
174.         }else{
175.             logger.error("client channel connect failed!");
176.             // 连接建立失败，连接关闭(上层catch住会关闭连接)
177.             throw new Error();
178.         }
179.     }
181.     private void processReadEvent(SelectionKey key) throws Exception {
182.         SocketChannel socketChannel = (SocketChannel)key.channel();
184.         // 简单起见，buffer不缓存，每次读事件来都新创建一个
185.         // 暂时也不考虑黏包/拆包场景(Netty中靠ByteToMessageDecoder解决，后续再分析其原理)，理想的认为每个消息都小于1024，且每次读事件都只有一个消息
186.         ByteBuffer readBuffer = ByteBuffer.allocate(64);
188.         int byteRead = socketChannel.read(readBuffer);
189.         if(byteRead == -1){
190.             // 简单起见不考虑tcp半连接的情况，返回-1直接关掉连接
191.             socketChannel.close();
192.             // 取消key的监听
193.             key.cancel();
194.         }else{
195.             // 将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作
196.             readBuffer.flip();
197.             // 根据缓冲区可读字节数创建字节数组
198.             byte[] bytes = new byte[readBuffer.remaining()];
199.             // 将缓冲区可读字节数组复制到新建的数组中
200.             readBuffer.get(bytes);
202.             if(myEventHandler != null) {
203.                 myEventHandler.fireChannelRead(socketChannel, bytes);
204.             }
205.         }
206.     }
208.     private void doRegister(SocketChannel socketChannel){
209.         try {
210.             // nio的非阻塞channel
211.             socketChannel.configureBlocking(false);
213.             socketChannel.finishConnect();
215.             logger.info("socketChannel={} finishConnect!",socketChannel);
217.             // 将接受到的连接注册到selector中，并监听read事件
218.             socketChannel.register(unwrappedSelector, SelectionKey.OP_READ);
220.             logger.info("socketChannel={} doRegister success!",socketChannel);
221.         }catch (Exception e){
222.             logger.error("register socketChannel={} error!",socketChannel,e);
223.             try {
224.                 socketChannel.close();
225.             } catch (IOException ex) {
226.                 logger.error("register channel close={} error!",socketChannel,ex);
227.             }
228.         }
229.     }
230. }

复制代码

MyNetty EventLoopGroup源码

[code]public class MyNioEventLoopGroup {    private final MyNioEventLoop[] executors;    private final int nThreads;    private final AtomicInteger atomicInteger = new AtomicInteger();    public MyNioEventLoopGroup(MyEventHandler myEventHandler, int nThreads) {        this(myEventHandler,nThreads,null);    }    public MyNioEventLoopGroup(MyEventHandler myEventHandler, int nThreads, MyNioEventLoopGroup childGroup) {        if(nThreads

鼓励转贴优秀软件安全工具和文档！