※推薦台中搬家公司優質服務,可到府估價
台中搬鋼琴,台中金庫搬運,中部廢棄物處理,南投縣搬家公司,好幫手搬家,西屯區搬家
前言
NioEventLoopGroup是netty對Reactor線程組這個抽象概念的具體實現，其內部維護了一個EventExecutor數組，而NioEventLoop就是EventExecutor的實現（看名字也可發現，一個是NioEventLoopGroup，一個是NioEventLoop，前者是集合，後者是集合中的元素）。一個NioEventLoop中運行着唯一的一個線程即Reactor線程，這個線程一直執行NioEventLoop的run方法。這個run方法就是netty的核心方法，其重要性可以類比於Spring中的refresh方法。
下面是從百度上隨便找的一篇netty文章的線程模型圖（詳見文章https://www.cnblogs.com/luoxn28/p/11875340.html），此處引用是為方便在頭腦中產生一個整體印象，結合圖下面的代碼進行各個概念的歸位。圖中綠色的Reactor Thread就是上文說的NioEventLoopGroup，對應下面代碼中的boss變量，負責處理客戶端的連接事件，它其實也是一個池（因為內部維護的是一個數組）；藍色的Reactor Thread Pool也是NioEventLoopGroup，對應下面代碼中的worker變量，負責處理客戶端的讀寫事件。
注：上圖是Reactor多線程模型，而下面的代碼示例是主從多線程模型，區別是只要將代碼boss中的參數2改成1，示例代碼就成了多線程模型，細細品味一下。
1 public class NettyDemo1 {
2 // netty服務端的一般性寫法
3 public static void main(String[] args) {
4 EventLoopGroup boss = new NioEventLoopGroup(2);
5 EventLoopGroup worker = new NioEventLoopGroup();
6 try {
7 ServerBootstrap bootstrap = new ServerBootstrap();
8 bootstrap.group(boss, worker).channel(NioServerSocketChannel.class)
9 .option(ChannelOption.SO_BACKLOG, 100)
10 .childHandler(new ChannelInitializer<SocketChannel>() {
11 @Override
12 protected void initChannel(SocketChannel socketChannel) throws Exception {
13 ChannelPipeline pipeline = socketChannel.pipeline();
14 pipeline.addLast(new StringDecoder());
15 pipeline.addLast(new StringEncoder());
16 pipeline.addLast(new NettyServerHandler());
17 }
18 });
19 ChannelFuture channelFuture = bootstrap.bind(90);
20 channelFuture.channel().closeFuture().sync();
21 } catch (Exception e) {
22 e.printStackTrace();
23 } finally {
24 boss.shutdownGracefully();
25 worker.shutdownGracefully();
26 }
27 }
28 }
以上部分是博主對netty的一個概括性總結，以將概念和其實現連接起來，方便建立一個初始的總體認識，下面進入EventLoopGroup的初始化。
一、EventLoopGroup初始化
1、NioEventLoopGroup構造器
順着有參和無參的構造方法進去，發現無參的構造器將線程數賦值0繼續調了有參的構造器，而有參的構造器將線程池executor參數賦值null繼續調重載構造器
1 public NioEventLoopGroup() {
2 this(0);
3 }
1 public NioEventLoopGroup(int nThreads) {
2 this(nThreads, (Executor) null);
3 }
1 public NioEventLoopGroup(int nThreads, Executor executor) {
2 this(nThreads, executor, SelectorProvider.provider());
3 }
因為博主是在筆記本電腦調試的，故此時的selectorProvider是WindowsSelectorProvider，然後又加了一個參數DefaultSelectStrategyFactory單例對象：
1 public NioEventLoopGroup(
2 int nThreads, Executor executor, final SelectorProvider selectorProvider) {
3 this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
4 }
然後調父類的構造器，在末尾增加一個參數RejectedExecutionHandler單例對象：
1 public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,
2 final SelectStrategyFactory selectStrategyFactory) {
3 super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
4 }
2、MultithreadEventLoopGroup構造器
在該構造器中，對線程數參數進行了處理，如果是0（對應上面NioEventLoopGroup的無參構造器），則將線程數設置為默認值，默認值取的是CPU核數*2，8核處理器對應16個線程；如果不是0，則以指定的線程數為準。同時，將executor後面的參數變為數組的形式，對應上面可以知道args中有三個元素：WindowsSelectorProvider、DefaultSelectStrategyFactory、RejectedExecutionHandler。
1 protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
2 super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
3 }
3、MultithreadEventExecutorGroup構造器
此構造器又在args數組前面加了一個單例對象DefaultEventExecutorChooserFactory，用於從NioEventLoopGroup的數組中選取一個NioEventLoop。
1 protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
2 this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
3 }
下面才是最終的核心構造器方法，結合註釋應該比較好理解。其中最重要的是第3步和第4步，下面着重講解這兩步。
1 protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
2 EventExecutorChooserFactory chooserFactory, Object... args) {
3 // 1.對線程數進行校驗
4 if (nThreads <= 0) {
5 throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
6 }
7 // 2.給線程池參數賦值，從前面追蹤可知，若未賦值，executor一直是null，後續用於創建NioEventLoop中的啟動線程，所以這玩意就是一個線程工廠
8 if (executor == null) {
9 executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
10 }
11 // 3.給children循環賦值，newChild方法是重點，後續會講解 ***
12 children = new EventExecutor[nThreads];
13 for (int i = 0; i < nThreads; i ++) {
14 boolean success = false;
15 try {
16 children[i] = newChild(executor, args);
17 success = true;
18 } catch (Exception e) {
19 // TODO: Think about if this is a good exception type
20 throw new IllegalStateException("failed to create a child event loop", e);
21 } finally {
22 // 省略掉未創建成功后的資源釋放處理
23 }
24 }
25 // 4.完成chooser選擇器的賦值，此處是netty一個小的優化點，後續會講解 **
26 chooser = chooserFactory.newChooser(children);
27 // 5.給數組中每一個成員設置監聽器處理
28 final FutureListener<Object> terminationListener = new FutureListener<Object>() {
29 @Override
30 public void operationComplete(Future<Object> future) throws Exception {
31 if (terminatedChildren.incrementAndGet() == children.length) {
32 terminationFuture.setSuccess(null);
33 }
34 }
35 };
36
37 for (EventExecutor e: children) {
38 e.terminationFuture().addListener(terminationListener);
39 }
40 // 6.設置一個只讀的set集合
41 Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
42 Collections.addAll(childrenSet, children);
43 readonlyChildren = Collections.unmodifiableSet(childrenSet);
44 }
3.1)、第4步chooser的賦值
由上面構造器調用過程可知，chooserFactory對應DefaultEventExecutorChooserFactory對象，該對象的newChooser方法如下：
1 public EventExecutorChooser newChooser(EventExecutor[] executors) {
2 if (isPowerOfTwo(executors.length)) {
3 return new PowerOfTwoEventExecutorChooser(executors);
4 } else {
5 return new GenericEventExecutorChooser(executors);
6 }
7 }
邏輯比較簡單，判斷數組的長度是不是2的N次冪，如果是，返回PowerOfTwoEventExecutorChooser對象，如果不是則返回GenericEventExecutorChooser對象。這二者有什麼區別，netty設計者為什麼要這麼做呢？如果對HashMap的實現原理有深入了解的園友應該不難想到，如果一個數X是2的N次冪，那麼用任意一個數Y對X取模可以用Y&(X-1)來高效的完成，這樣做比直接%取模快了好幾倍，這也是HashMap用2次冪作為數組長度的主要原因。這裡是同樣的道理，如下代碼所示，這兩個chooser類都很簡單，內部維護了一個原子遞增對象，每次調用next方法都加1，然後用這個數與數組長度取模，得到要對應下標位置的元素。而如果數組長度剛好是2次冪，用PowerOfTwoEventExecutorChooser就會提高效率，如果不是那也沒辦法，走%取模就是了。netty這種對效率提升的處理，是否在平時的CRUD中也能套用一下呢？
1 private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {
2 private final AtomicInteger idx = new AtomicInteger();
3 private final EventExecutor[] executors;
4
5 PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {
6 this.executors = executors;
7 }
8
9 @Override
10 public EventExecutor next() {
11 return executors[idx.getAndIncrement() & executors.length - 1];
12 }
13 }
14
15 private static final class GenericEventExecutorChooser implements EventExecutorChooser {
16 private final AtomicInteger idx = new AtomicInteger();
17 private final EventExecutor[] executors;
18
19 GenericEventExecutorChooser(EventExecutor[] executors) {
20 this.executors = executors;
21 }
22
23 @Override
24 public EventExecutor next() {
25 return executors[Math.abs(idx.getAndIncrement() % executors.length)];
26 }
27 }
※台中搬家公司教你幾個打包小技巧,輕鬆整理裝箱!
還在煩惱搬家費用要多少哪？台中大展搬家線上試算搬家費用，從此不再擔心「物品怎麼計費」、「多少車才能裝完」
3.2)、第3步newChild方法的邏輯
該方法的實現在NioEventLoopGroup中，由於args長度為3，所以queueFactory為null（暫時未發現哪裡的實現args參數長度會是4，或許只是為後續擴展用，如果園友對args長度為4的場景有了解的還請留言指教）。然後調用了NioEventLoop的構造器，下面進入NioEventLoop的初始化。
1 protected EventLoop newChild(Executor executor, Object... args) throws Exception {
2 EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;
3 return new NioEventLoop(this, executor, (SelectorProvider) args[0],
4 ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2], queueFactory);
5 }
執行完上述初始化方法后NioEventLoopGroup的快照圖如下，最重要的就兩個屬性：child和chooser。
二、NioEventLoop的初始化
1、NioEventLoop的構造器
到這裏，有必要將此構造器的入參再梳理一遍。parent即上面的NioEventLoopGroup對象，executor是在MultithreadEventExecutorGroup中初始化的ThreadPerTaskExecutor，selectorProvider是WindowsSelectorProvider，strategy是DefaultSelectStrategyFactory，rejectedExecutionHandler是RejectedExecutionHandler，queueFactory是null。
1 NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
2 SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
3 EventLoopTaskQueueFactory queueFactory) {
4 super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
5 rejectedExecutionHandler);
6 if (selectorProvider == null) {
7 throw new NullPointerException("selectorProvider");
8 }
9 if (strategy == null) {
10 throw new NullPointerException("selectStrategy");
11 }
12 provider = selectorProvider;
13 final SelectorTuple selectorTuple = openSelector();
14 selector = selectorTuple.selector;// netty封裝的selector
15 unwrappedSelector = selectorTuple.unwrappedSelector;// java NIO原生的selector
16 selectStrategy = strategy;
17 }
可以看到只是做了一些賦值，其中newTaskQueue方法創建的是MpscUnboundedArrayQueue隊列（多生產單消費無界隊列，mpsc是multi provider single consumer的首字母縮寫，即多個生產一個消費），繼續追查父類構造方法。
2、SingleThreadEventLoop構造器
調用父類構造器，給tailTasks賦值。
1 protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,
2 boolean addTaskWakesUp, Queue<Runnable> taskQueue, Queue<Runnable> tailTaskQueue,
3 RejectedExecutionHandler rejectedExecutionHandler) {
4 super(parent, executor, addTaskWakesUp, taskQueue, rejectedExecutionHandler);
5 tailTasks = ObjectUtil.checkNotNull(tailTaskQueue, "tailTaskQueue");
6 }
3、SingleThreadEventExecutor構造器
在該構造方法中完成了剩餘變量的賦值，其中有兩個變量很重要：executor和taskQueue。前者負責創建Reactor線程，後者是實現串行無鎖化的任務隊列。
1 protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
2 boolean addTaskWakesUp, Queue<Runnable> taskQueue,
3 RejectedExecutionHandler rejectedHandler) {
4 super(parent);
5 this.addTaskWakesUp = addTaskWakesUp;
6 this.maxPendingTasks = DEFAULT_MAX_PENDING_EXECUTOR_TASKS;
7 this.executor = ThreadExecutorMap.apply(executor, this);
8 this.taskQueue = ObjectUtil.checkNotNull(taskQueue, "taskQueue");
9 rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
10 }
NioEventLoopGroup的對象引用最終記錄在了AbstractEventExecutor中：
1 protected AbstractEventExecutor(EventExecutorGroup parent) {
2 this.parent = parent;
3 }
NioeventLoop初始化完成之後的對象快照如下，左邊是子類，右邊是父類：
小結
本文詳細講述了netty中Reactor線程組概念模型的實現類 -- NioEventLoopGroup的實例化過程。NioEventLoopGroup和其內部數組元素NioEventLoop是netty通信框架的基石，相信本文的內容對初學netty的園友有一點幫助。
下篇將研究ServerBootstrap的初始化過程，敬請期待。
本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理
※台中搬家公司教你幾個打包小技巧,輕鬆整理裝箱!
還在煩惱搬家費用要多少哪？台中大展搬家線上試算搬家費用，從此不再擔心「物品怎麼計費」、「多少車才能裝完」
Orignal From: Netty源碼學習系列之2-NioEventLoopGroup的初始化_台中搬家公司
留言
張貼留言