Netty服务端的业务流程分析

2023-01-05 14:19:52

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
 
 
 
 
 
上一篇：IsWindow,findwindow


下一篇：ASP.NET MVC扩展自定义视图引擎支持多模板&动态换肤skins机制