1. 程式人生 > >Netty原始碼分析:1.1初始化NioEventLoopGroup

Netty原始碼分析:1.1初始化NioEventLoopGroup

第一章節是主要是伺服器啟動的程式碼分析。章節目錄有:
|———1.1初始化NioEventLoopGroup
|———1.2初始化NioEventLoop
|———1.3初始化NioServerSocketChannel
|———1.4伺服器啟動流程
為什麼先從初始化開始瞭解伺服器啟動?
因為在我看伺服器啟動的相關原始碼的時候,有很多地方都是初始化的時候已經建立好的。所以我就從初始化的原始碼開始看起。這是我第一次看原始碼的筆記,仍有很多理解錯誤的地方和不解的地方。歡迎討論。

本篇目錄:

  • 繼承關係類圖
  • 初始化時序圖
  • NioEventLoopGroup原始碼分析
  • 總結

繼承關係類圖

image.png

初始化時序圖

image.png

程式碼debug開始於new NioEventLoopGroup();

1. NioEventLoopGroup預設有參構造

雖然我們在new NioEventLoopGroup()的時候並沒有傳入引數,但是在初始化的時候,NioEventLoopGroup 類會自動呼叫自己的有參建構函式,並且呼叫父類的有參構造進行初始化。
- 第一個引數int nThreads: 建立多少個執行緒數,如果傳入是0.則預設是cpu*2的個數。
- 第二個引數Executor executor: 初始化是null。
- 第三個引數SelectorProvider selectorProvider

:是一個SelectorProvider.provider()的返回值。
- 第四個引數RejectedExecutionHandler的返回值:是一個處理異常的類

SelectorProvider就是為了建立DatagramChannel,Pipe,Selector,ServerSocketChannel,SocketChannel,System.inheritedChannel()而服務的

public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,
                             final
SelectStrategyFactory selectStrategyFactory) { super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject()); }

2. MultithreadEventLoopGroup

接下來是父類的初始化.在這一步可以看到是繼續呼叫super()進行一個父類的有參構造。 在這裡可以看到nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads .也就是說如果new NioEventLoopGroup() 的時候沒有定義nThreads那麼預設值是0,如果是0,則傳入的數是DEFAULT_EVENT_LOOP_THREADS(cpu*2的執行緒數。)

 protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }

3. MultithreadEventExecutorGroup

在這裡 MultithreadEventExecutorGroup 再一次呼叫自己有參構造,並傳入一個引數,DefaultEventExecutorChooserFactory.INSTANCE,其實是返回一個DefaultEventExecutorChooserFactory物件。該物件的作用就是獲取EventExecutor物件的。

DefaultEventExecutorChooserFactory的內部原理就是:呼叫newChooser(EventExecutor[] e)方法,傳入EventExecutor陣列,返回一個PowerOfTwoEventExecutorChooser物件,該物件作用就是呼叫next()方法獲取一個EventExecutor物件。所以說DefaultEventExecutorChooserFactory類就是獲取EventExecutor物件的。

protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
        this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
    }

接著是呼叫MultithreadEventExecutorGroup 另一個有參構造。
- 1 判斷executor 是否為空,為空就建立一個ThreadPerTaskExecutor物件。
ThreadPerTaskExecutor
實現executor介面。
- 2 建立一個EventExecutor陣列個數為nThreads個
- 3 初始化EventExecutor陣列。呼叫的是newChild方法.下面分析一下newChild方法.步驟是:建立一個NioEventLoop執行緒,如果建立不成功
- 4 finally 是判斷是否建立NioEventLoop物件成功。如果失敗,就關閉執行緒。
- 5 在初始化第3步提到,就是用DefaultEventExecutorChooserFactory物件 管理一個EventExecutor陣列。
- 6 建立一個監聽器,當操作完成的時候,設定一個成功的資訊。
- 7 給每一個EventExecutor新增一個監聽器
- 8 建立一個set 存放這個EventExecutor陣列。並賦予readonlyChildren變數

到此,初始化就已經完成了。

 protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                            EventExecutorChooserFactory chooserFactory, Object... args) {

        //檢驗引數
        if (nThreads <= 0) {
            throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
        }

        //1
        if (executor == null) {
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }


        //2  
        children = new EventExecutor[nThreads];

        //3
        for (int i = 0; i < nThreads; i ++) {
            boolean success = false;
            try {
                children[i] = newChild(executor, args);
                success = true;
            } catch (Exception e) {

                throw new IllegalStateException("failed to create a child event loop", e);
            } finally {
               //4
                        }
                    }
                }
            }
        }

        //5
        chooser = chooserFactory.newChooser(children);

        //6
        final FutureListener<Object> terminationListener = new FutureListener<Object>() {
            @Override
            public void operationComplete(Future<Object> future) throws Exception {
                if (terminatedChildren.incrementAndGet() == children.length) {
                    terminationFuture.setSuccess(null);
                }
            }
        };

        //7
        for (EventExecutor e: children) {
            e.terminationFuture().addListener(terminationListener);
        }
        //8
        Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
        Collections.addAll(childrenSet, children);
        readonlyChildren = Collections.unmodifiableSet(childrenSet);
    }

分析newChild()

在這裡NioEventLoopGroup重寫了父類MultithreadEventExecutorGroup的newChild()方法
可以看出是建立一個NioEventLoop的物件。

@Override
    protected EventLoop newChild(Executor executor, Object... args) throws Exception {
        return new NioEventLoop(this, executor, (SelectorProvider) args[0],
            ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
    }

總結

我覺得NioEventLoopGroup的初始化就是,建立預設cup核數*2個的NioEventLoop。當有一個channel新連線的時候,將該channel與一個NioEventLoop繫結,該channel的任務全部交給NioEventLoop執行了。