nioeventloopgroup 多少比较好

Posted 2023-04-06

参考技术A 相比Netty3,Netty4有很多显著的变化：NioEventLoopGroup是一个处理I/O操作的多线程事件环。即为Netty4里的线程池，在3.x里，一个Channel是由ChannelFactory创建的，同时新创建的Channel会自动注册到一个隐藏的I/O线程。4.0使用新的名为EventLoopGroup的接口来替换ChannelFactory，它由一个或多个EventLoop来构成。一个新的Channel不会自动注册到EventLoopGroup，但用户可以显式调用EventLoopGroup.register（）来注册。在Server端的Bootstrap参数中，有两个EventLoopGroup，第一个通常称为'boss'，用于接收发来的连接请求。第二个称为'worker',，用于处理boss接受并且注册给worker的连接中的信息。ChannelInitializer是一个特殊的handler，用于方便的配置用户自定义的handler实现，如代码中所示。在channelRegistered的生命周期中会触发用户复写的initChannel(Cch)方法，并且在调用后会讲自身从channelPipeline中移除。代码示例?importio.netty.bootstrap.ServerBootstrap;importio.netty.channel.ChannelFuture;importio.netty.channel.ChannelHandlerContext;importio.netty.channel.ChannelInboundHandlerAdapter;importio.netty.channel.ChannelInitializer;importio.netty.channel.ChannelOption;importio.netty.channel.ChannelPipeline;importio.netty.channel.EventLoopGroup;importio.netty.channel.nio.NioEventLoopGroup;importio.netty.channel.socket.SocketChannel;importio.netty.channel.socket.nio.NioserverSocketChannel;importio.netty.handler.codec.serialization.ClassResolvers;importio.netty.handler.codec.serialization.ObjectDecoder;importio.netty.handler.codec.serialization.ObjectEncoder;importio.netty.handler.logging.LogLevel;importio.netty.handler.logging.LoggingHandler;importio.netty.handler.ssl.SslContext;importio.netty.handler.ssl.util.SelfSignedCertificate;/***DateTime:2015年1月5日上午9:56:10**/publicclassHelloWorldServerstaticfinalbooleanSSL=System.getProperty("ssl")!=null;staticfinalintPORT=Integer.parseInt(System.getProperty("port","8007"));publicstaticvoidmain(String[]args)throwsException//ConfigureSSL.finalSslContextsslCtx;if(SSL)SelfSignedCertificatessc=newSelfSignedCertificate();sslCtx=SslContext.newServerContext(ssc.certificate(),ssc.privateKey());elsesslCtx=null;//Configuretheserver.EventLoopGroupbossGroup=newNioEventLoopGroup(1);EventLoopGroupworkerGroup=newNioEventLoopGroup();tryServerBootstrapb=newServerBootstrap();b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,100).handler(newLoggingHandler(LogLevel.INFO)).childHandler(newChannelInitializer()@OverridepublicvoidinitChannel(SocketChannelch)throwsExceptionChannelPipelinep=ch.pipeline();if(sslCtx!=null)p.addLast(sslCtx.newHandler(ch.alloc()));p.addLast(newLoggingHandler(LogLevel.INFO));p.addLast(newObjectEncoder(),newObjectDecoder(ClassResolvers.cacheDisabled(null)),newHelloWorldServerHandler()););//Starttheserver.ChannelFuturef=b.bind(PORT).sync();//Waituntiltheserversocketisclosed.f.channel().closeFuture().sync();finally//Shutdownalleventloopstoterminateallthreads.bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();classHelloWorldServerHandlerextendsChannelInboundHandlerAdapter@OverridepublicvoidchannelRead(ChannelHandlerContextctx,Objectmsg)ctx.write("serverwritemsg:"+msg);@OverridepublicvoidchannelReadComplete(ChannelHandlerContextctx)ctx.flush();@OverridepublicvoidexceptionCaught(ChannelHandlerContextctx,Throwablecause)//Closetheconnectionwhenanexceptionisraised.cause.printStackTrace();ctx.close();importio.netty.bootstrap.Bootstrap;importio.netty.channel.ChannelFuture;importio.netty.channel.ChannelHandlerContext;importio.netty.channel.ChannelInboundHandlerAdapter;importio.netty.channel.ChannelInitializer;importio.netty.channel.ChannelOption;importio.netty.channel.ChannelPipeline;importio.netty.channel.EventLoopGroup;importio.netty.channel.nio.NioEventLoopGroup;importio.netty.channel.socket.SocketChannel;importio.netty.channel.socket.nio.NioSocketChannel;importio.netty.handler.codec.serialization.ClassResolvers;importio.netty.handler.codec.serialization.ObjectDecoder;importio.netty.handler.codec.serialization.ObjectEncoder;importio.netty.handler.ssl.SslContext;importio.netty.handler.ssl.util.InsecureTrustManagerFactory;/***DateTime:2015年1月5日上午9:56:22**/publicclassHelloWorldClientstaticfinalbooleanSSL=System.getProperty("ssl")!=null;staticfinalStringHOST=System.getProperty("host","127.0.0.1");staticfinalintPORT=Integer.parseInt(System.getProperty("port","8007"));publicstaticvoidmain(String[]args)throwsException//ConfigureSSL.gitfinalSslContextsslCtx;if(SSL)sslCtx=SslContext.newClientContext(InsecureTrustManagerFactory.INSTANCE);elsesslCtx=null;//Configuretheclient.EventLoopGroupgroup=newNioEventLoopGroup();tryBootstrapb=newBootstrap();b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).handler(newChannelInitializer()@OverridepublicvoidinitChannel(SocketChannelch)throwsExceptionChannelPipelinep=ch.pipeline();if(sslCtx!=null)p.addLast(sslCtx.newHandler(ch.alloc(),HOST,PORT));p.addLast(newObjectEncoder(),newObjectDecoder(ClassResolvers.cacheDisabled(null)),newHelloWorldClientHandler()););//Starttheclient.ChannelFuturef=b.connect(HOST,PORT).sync();//Waituntiltheconnectionisclosed.f.channel().closeFuture().sync();finally//Shutdowntheeventlooptoterminateallthreads.group.shutdownGracefully();classHelloWorldClientHandlerextendsChannelInboundHandlerAdapterprivatefinalStringmsg="hellojavaworld";/***Createsaclient-sidehandler.*/publicHelloWorldClientHandler()//TODO@OverridepublicvoidchannelActive(ChannelHandlerContextctx)ctx.writeAndFlush(msg);@OverridepublicvoidchannelRead(ChannelHandlerContextctx,Objectmsg)System.out.println(msg);//ctx.write(msg);@OverridepublicvoidchannelReadComplete(ChannelHandlerContextctx)ctx.flush();@OverridepublicvoidexceptionCaught(ChannelHandlerContextctx,Throwablecause)//Closetheconnectionwhenanexceptionisraised.cause.printStackTrace();ctx.close();

Netty4.XNetty源码分析之NioEventLoopGroup

NioEventLoopGroup实际是NioEventLoop的线程组，主要负责管理EventLoop的生命周期，EventLoop的默认大小是2倍的CPU核数，但这并不是一个恒定的最佳数量，为了避免线程上下文切换，只要能满足要求，这个值其实越小越好。

继承关系如下：

首先看NioEventLoopGroup构造方法：

public NioEventLoopGroup() 
    this(0);


public NioEventLoopGroup(int nThreads) 
    this(nThreads, null);


public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory) 
    this(nThreads, threadFactory, SelectorProvider.provider());


public NioEventLoopGroup(
        int nThreads, ThreadFactory threadFactory, final SelectorProvider selectorProvider) 
    super(nThreads, threadFactory, selectorProvider);

从以上代码可以发现NioEventLoopGroup虽然有4个构造方法，但最终调用的是MultithreadEventLoopGroup的构造方法，代码如下：

protected MultithreadEventLoopGroup(int nThreads, ThreadFactory threadFactory, Object... args)   
    super(nThreads == 0? DEFAULT_EVENT_LOOP_THREADS : nThreads, threadFactory, args);  


private static final int DEFAULT_EVENT_LOOP_THREADS;

    static 
        DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
                "io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));

        if (logger.isDebugEnabled()) 
            logger.debug("-Dio.netty.eventLoopThreads: ", DEFAULT_EVENT_LOOP_THREADS);
        
    

在NioEventLoopGroup初始化之前，会先执行父类MultithreadEventLoopGroup的静态模块，其中DEFAULT_EVENT_LOOP_THREADS的大小默认是CPU核数的两倍，就是文章开头说过的。

MultithreadEventExecutorGroup的构造方法：

//EventExecutor数组，保存eventLoop
private final EventExecutor[] children;
//从children中选取一个eventLoop的策略
private final EventExecutorChooser chooser;

protected MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory, Object... args) 
    if (nThreads <= 0) 
        throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
    

    if (threadFactory == null) 
        //是一个通用的ThreadFactory实现，方便配置线程池
        threadFactory = newDefaultThreadFactory();
    

    //根据线程数来创建SingleThreadEventExecutor数组
    //从命名上可以看出SingleThreadEventExecutor是一个只有一个线程的线程池
    children = new SingleThreadEventExecutor[nThreads];
    //根据children数组的大小，采用不同策略初始化chooser。
    if (isPowerOfTwo(children.length)) 
        chooser = new PowerOfTwoEventExecutorChooser();
     else 
        chooser = new GenericEventExecutorChooser();
    

    for (int i = 0; i < nThreads; i ++) 
        boolean success = false;
        try 
            //
            children[i] = newChild(threadFactory, args);
            success = true;
         catch (Exception e) 
            // TODO: Think about if this is a good exception type
            throw new IllegalStateException("failed to create a child event loop", e);
         finally 
            //如果没有创建成功，循环关闭所有SingleThreadEventExecutor
            if (!success) 
                for (int j = 0; j < i; j ++) 
                    children[j].shutdownGracefully();
                

                //等待关闭成功
                for (int j = 0; j < i; j ++) 
                    EventExecutor e = children[j];
                    try 
                        while (!e.isTerminated()) 
                            e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
                        
                     catch (InterruptedException interrupted) 
                        Thread.currentThread().interrupt();
                        break;
                    
                
            
        
    

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

    for (EventExecutor e: children) 
        e.terminationFuture().addListener(terminationListener);
    

回到NioEventLoopGroup的newChild方法重载

@Override
protected EventExecutor newChild(
        ThreadFactory threadFactory, Object... args) throws Exception 
    return new NioEventLoop(this, threadFactory, (SelectorProvider) args[0]);

MultithreadEventExecutorGroup构造方法中执行的是NioEventLoopGroup中的newChild方法，所以children元素的实际类型是NioEventLoop。

解释下EventExecutorChooser的选择

    //判断一个数是否是2的幂次方
    private static boolean isPowerOfTwo(int val) 
        return (val & -val) == val;
    

    private final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser 
        @Override
        public EventExecutor next() 
            return children[childIndex.getAndIncrement() & children.length - 1];
        
    

    private final class GenericEventExecutorChooser implements EventExecutorChooser 
        @Override
        public EventExecutor next() 
            return children[Math.abs(childIndex.getAndIncrement() % children.length)];
        
    

它是根据线程数组大小是否是2的幂次方来选择初始化chooser。如果大小为2的幂次方，则采用PowerOfTwoEventExecutorChooser，否则使用GenericEventExecutorChooser。也就是如果线程数是2的倍数时，Netty选择线程时会使用PowerOfTwoEventExecutorChooser，因为&比%更快(Netty为了性能也是拼了)。

作者：小毛驴，一个游戏人　
梦想：世界和平　　　
github主页：https://liulongling.github.io/
csdn主页：http://blog.csdn.net/liulongling
简书主页:http://www.jianshu.com/u/1342f21e1748
若有错误之处，请多多谅解并欢迎批评指正。　　　　
本博客中未标明转载的文章归作者小毛驴所有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。