Netty服務端的業務流程分析

Netty的服務端怎麼和java NIO聯繫起來的，一直很好奇這塊內容，這裏跟下代碼，下篇文章看下Channel相關的知識。

1. finalChannelFuture initAndRegister(){
2. finalChannel channel = channelFactory().newChannel();//
3. try{
4. init(channel);
5. }catch(Throwable t){
6. channel.unsafe().closeForcibly();//當即關閉通道且不會觸發事件
7. //由於這個通道尚未註冊到EventLoop，因此咱們須要強制GlobalEventExecutor的使用。
8. returnnewDefaultChannelPromise(channel,GlobalEventExecutor.INSTANCE).setFailure(t);
9. }
10. //註冊一個EventLoop
11. ChannelFuture regFuture = group().register(channel);
12. //註冊失敗
13. if(regFuture.cause()!=null){
14. if(channel.isRegistered()){
15. channel.close();
16. }else{
17. channel.unsafe().closeForcibly();
18. }
19. }
21. // If we are here and the promise is not failed, it's one of the following cases:
22. // 程序運行到這裏且promise沒有失敗，可能有以下幾種狀況
23. // 1) If we attempted registration from the event loop, the registration has been completed at this point.
24. // 若是試圖註冊到一個EventLoop,該註冊完成，
25. // i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
26. // 2) If we attempted registration from the other thread, the registration request has been successfully
27. // added to the event loop's task queue for later execution.
28. // 若是試圖註冊到其餘線程，該註冊已經成功，可是沒有完成，添加一個事件到任務隊列中，等會執行
29. // i.e. It's safe to attempt bind() or connect() now:
30. // because bind() or connect() will be executed *after* the scheduled registration task is executed
31. // because register(), bind(), and connect() are all bound to the same thread.
33. return regFuture;
34. }

注意這裏的Channle的類型爲NioServerSocketChannel類型，group()返回NioEventLoopGroup類型，他繼承 MultithreadEventLoopGroup， 那麼看下register的實現：

1. @Override
2. publicChannelFutureregister(Channel channel){
3. return next().register(channel);
4. }

跟進去是調用了SingleThreadEventLoop類的register方法。實現以下.

1. @Override
2. publicChannelFutureregister(finalChannel channel,finalChannelPromise promise){
3. if(channel ==null){
4. thrownewNullPointerException("channel");
5. }
6. if(promise ==null){
7. thrownewNullPointerException("promise");
8. }
10. channel.unsafe().register(this, promise);
11. return promise;
12. }

調用了 NioServerSocketChannel的unsafe()的register方法。

1. @Override
2. publicfinalvoidregister(EventLoop eventLoop,finalChannelPromise promise){
3. if(eventLoop ==null){
4. thrownewNullPointerException("eventLoop");
5. }
6. if(isRegistered()){
7. promise.setFailure(newIllegalStateException("registered to an event loop already"));
8. return;
9. }
10. if(!isCompatible(eventLoop)){
11. promise.setFailure(
12. newIllegalStateException("incompatible event loop type: "+ eventLoop.getClass().getName()));
13. return;
14. }
16. AbstractChannel.this.eventLoop = eventLoop;
18. if(eventLoop.inEventLoop()){
19. register0(promise);
20. }else{
21. try{
22. eventLoop.execute(newOneTimeTask(){
23. @Override
24. publicvoid run(){
25. register0(promise);
26. }
27. });
28. }catch(Throwable t){
29. logger.warn(
30. "Force-closing a channel whose registration task was not accepted by an event loop: {}",
31. AbstractChannel.this, t);
32. closeForcibly();
33. closeFuture.setClosed();
34. safeSetFailure(promise, t);
35. }
36. }
37. }

對這個eventloop.inEventLoop的理解不是很深入，有點像android開發裏面費時的操做不要放到主線程裏面。 eventLoop . inEventLoop ()表示不在主線程裏面。

register的最終實如今，AbstractNioChannel類裏面：

1. @Override
2. protectedvoid doRegister()throwsException{
3. boolean selected =false;
4. for(;;){
5. try{
6. selectionKey = javaChannel().register(eventLoop().selector,0,this);
7. return;
8. }catch(CancelledKeyException e){
9. if(!selected){
10. // Force the Selector to select now as the "canceled" SelectionKey may still be
11. // cached and not removed because no Select.select(..) operation was called yet.
12. eventLoop().selectNow();
13. selected =true;
14. }else{
15. // We forced a select operation on the selector before but the SelectionKey is still cached
16. // for whatever reason. JDK bug ?
17. throw e;
18. }
19. }
20. }
21. }

javaChannel()返回Java的Channel對象，eventLoop()返回NioEventLoop對象。裏面包含一個selector對象。selectNow是個非阻塞的調用， 調用此方法會清除全部之前調用  wakeup  方法所得的結果

Netty的Channel是對JDK中Channel的包裝和擴展。

 

註冊成功後就須要綁定端口了，

1. privatestaticvoid doBind0(
2. finalChannelFuture regFuture,finalChannel channel,
3. finalSocketAddress localAddress,finalChannelPromise promise){
5. // This method is invoked before channelRegistered() is triggered. Give user handlers a chance to set up
6. // the pipeline in its channelRegistered() implementation.
7. channel.eventLoop().execute(newRunnable(){
8. @Override
9. publicvoid run(){
10. if(regFuture.isSuccess()){
11. channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
12. }else{
13. promise.setFailure(regFuture.cause());
14. }
15. }
16. });
17. }

除了監聽端口還不夠，還要處理IO事件：

1. @Override
2. protectedvoid run(){
3. for(;;){
4. boolean oldWakenUp = wakenUp.getAndSet(false);
5. try{
6. if(hasTasks()){
7. selectNow();
8. }else{
9. select(oldWakenUp);
11. // 'wakenUp.compareAndSet(false, true)' is always evaluated
12. // before calling 'selector.wakeup()' to reduce the wake-up
13. // overhead. (Selector.wakeup() is an expensive operation.)
14. //
15. // However, there is a race condition in this approach.
16. // The race condition is triggered when 'wakenUp' is set to
17. // true too early.
18. //
19. // 'wakenUp' is set to true too early if:
20. // 1) Selector is waken up between 'wakenUp.set(false)' and
21. // 'selector.select(...)'. (BAD)
22. // 2) Selector is waken up between 'selector.select(...)' and
23. // 'if (wakenUp.get()) { ... }'. (OK)
24. //
25. // In the first case, 'wakenUp' is set to true and the
26. // following 'selector.select(...)' will wake up immediately.
27. // Until 'wakenUp' is set to false again in the next round,
28. // 'wakenUp.compareAndSet(false, true)' will fail, and therefore
29. // any attempt to wake up the Selector will fail, too, causing
30. // the following 'selector.select(...)' call to block
31. // unnecessarily.
32. //
33. // To fix this problem, we wake up the selector again if wakenUp
34. // is true immediately after selector.select(...).
35. // It is inefficient in that it wakes up the selector for both
36. // the first case (BAD - wake-up required) and the second case
37. // (OK - no wake-up required).
39. if(wakenUp.get()){
40. selector.wakeup();
41. }
42. }
44. cancelledKeys =0;
45. needsToSelectAgain =false;
46. finalint ioRatio =this.ioRatio;
47. if(ioRatio ==100){
48. processSelectedKeys();
49. runAllTasks();
50. }else{
51. finallong ioStartTime =System.nanoTime();
53. processSelectedKeys();
55. finallong ioTime =System.nanoTime()- ioStartTime;
56. runAllTasks(ioTime *(100- ioRatio)/ ioRatio);
57. }
59. if(isShuttingDown()){
60. closeAll();
61. if(confirmShutdown()){
62. break;
63. }
64. }
65. }catch(Throwable t){
66. logger.warn("Unexpected exception in the selector loop.", t);
68. // Prevent possible consecutive immediate failures that lead to
69. // excessive CPU consumption.
70. try{
71. Thread.sleep(1000);
72. }catch(InterruptedException e){
73. // Ignore.
74. }
75. }
76. }
77. }

該方法是NioEventLoop

 

 

 

 

 

來自爲知筆記(Wiz)