教你正确地利用Netty建立连接池

Posted 2021-04-06 欺世万宁人

一、问题描述

Netty是最近非常流行的高性能异步通讯框架，相对于Java原生的NIO接口，Netty封装后的异步通讯机制要简单很多。

但是小K最近发现并不是所有开发人员在使用的过程中都了解其内部实现机制，而是照着葫芦画瓢。

网上简单搜索下，在客户端使用Netty建立连接池的文章也是比较少。今天小K给大家简单介绍下使用Netty建立连接池的方法。

首先我们来看下Netty官方给出的客户端sample实例：

  //创建一个EventLoopGroup，可以简单认为是Netty框架下的线程池，默认最大线程数量是处理器数量的2倍
  EventLoopGroup group = new NioEventLoopGroup();
  try {
//Netty建立连接的辅助类
      Bootstrap b = new Bootstrap();
//配置属性，向pipeline添加handler
      b.group(group)
       .channel(NiosocketChannel.class)
       .option(ChannelOption.TCP_NODELAY, true)
       .handler(new ChannelInitializer<SocketChannel>() {
           @Override
           public void initChannel(SocketChannel ch) throws Exception {
               ChannelPipeline p = ch.pipeline();
               if (sslCtx != null) {
                   p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
               }
               //p.addLast(new LoggingHandler(LogLevel.INFO));
               p.addLast(new EchoClientHandler());
           }
       });

      //启动建立连接
      ChannelFuture f = b.connect(HOST, PORT).sync();

      //block直到连接被关闭
      f.channel().closeFuture().sync();

很简单？没错，确实如此。那么现在问题来了，如果你现在需要连接100个服务器，你会怎么做呢？

下面这样处理怎么样呢？我们在外层加了一个for循环

for(Host host : hosts){
          //创建一个EventLoopGroup，可以简单认为是Netty框架下的线程池，默认线程数量是处理器数量的2倍
          EventLoopGroup group = new NioEventLoopGroup();
          try {
　　　　　　　　//Netty建立连接的辅助类
              Bootstrap b = new Bootstrap();
　　　　　　　　//配置属性，向pipeline添加handler
              b.group(group)
               .channel(NioSocketChannel.class)
               .option(ChannelOption.TCP_NODELAY, true)
               .handler(new ChannelInitializer<SocketChannel>() {
                   @Override
                   public void initChannel(SocketChannel ch) throws Exception {
                       ChannelPipeline p = ch.pipeline();
                       if (sslCtx != null) {
                           p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
                       }
                       //p.addLast(new LoggingHandler(LogLevel.INFO));
                       p.addLast(new EchoClientHandler());
                   }
               });

              //启动建立连接
              ChannelFuture f = b.connect(HOST, PORT).sync();

              //block直到连接被关闭
              f.channel().closeFuture().sync();
}

问题很明显，如果每一个channel都对应一个NIOEventLoopGroup，那么我们实际上构建了一个connection:thread = 1:1的模型，随着连接数不断地扩大，线程膨胀的问题就会突显出来。

二、问题解决

那么如何避免线程膨胀的问题呢？很简单，我们只要稍微修改下上面的代码就可以了。

NioEventLoopGroup group = new NioEventLoopGroup();
    Bootstrap b = new Bootstrap();

    try {
      b.group(group)
       .channel(NioSocketChannel.class)
       .option(ChannelOption.TCP_NODELAY, true)
       .handler(new ChannelInitializer<SocketChannel>() {
           @Override
           public void initChannel(SocketChannel ch) throws Exception {
               ChannelPipeline p = ch.pipeline();
               if (sslCtx != null) {
                   p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
               }
               //p.addLast(new LoggingHandler(LogLevel.INFO));
                           p.addLast(new EchoClientHandler());
                       }
         });
     for(Host HOST : Hosts){
         ChannelFuture f = b.connect(HOST, PORT).sync();
     }

在上面的代码中，我们使用同一个创建了多个连接，从而使连接共享了一个NioEventLoopGroup，避免了线程膨胀的问题。

问题就这样解决了吗？NO，还远远没有结束哦，那么问题又来了。

1、如果希望每个连接能够使用不同的Handler怎么办？

2、每个连接如何能够再次复用，避免重复创建channel？

为了能够创建异步操作的连接池我们需要实现如下的模型。

为了能够方便地建立一个异步操作的连接池，我们会使用到FixedChannelPool（不了解的同学麻烦Google一下吧，（⌒_⌒））

其伪代码如下(具体的代码实现结构还是留给读者自己思考吧)：

Bootstrap b = new Bootstrap().channel(NioSocketChannel.class).group(
new NioEventLoopGroup());

//自定义的channelpoolhandler
ChannelPoolHandler handler = new ChannelPoolHandler();

//创建一个FixedChannelPool
FixedChannelPool pool = new FixedChannelPool(bootstrap, handler);

//从channelpool中获取连接或者创建新的连接，并添加listener
pool.acquire.addlistener();

三、总结

通常情况下，我们并不需要使用Netty建立连接池，common pool可以满足我们的需求，但是有些业务场景（例如：返回结果时间不确定）需要使用这种异步的连接池，在正确的业务场景下选择正确的解决方案才是王道哦。

来源于码农网