前言

目前的RPC框架中消费者端是调用一次服务就会进行一些netty的初始化,绑定端口等操作。那发起100次的话就是进行100次这样的连接，消耗的资源太多了。所以希望客户端池化。

Netty线程池

之前自己是初步接触Netty，用的还是5.x版本。再查阅了资料发现Netty自带了线程池很开心，但是自己试了半天好像没有这个工具类，才发现好像4.x版本是有的，一查发现5.x版本好像弃用了，我的天只好降版本了。（不知道为什么5.x版本弃用了）

废话不多说直接贴代码（把客户端改造了一下可以看我之前写的自己设计一个的轻量级的RPC框架–客户端netty 对比一些）
注意初始化完毕之后我们去FixedChannelPool中获取的channelhandler的时候用完需要放回去。

public class RPCRequestNet { 
    //管理以ip:端口号为key的连接池 FixedChannelPool继承SimpleChannelPool，有大小限制的连接池实现
    public static ChannelPoolMap<InetSocketAddress, FixedChannelPool> poolMap;
    //启动辅助类 用于配置各种参数
    private static Bootstrap b=new Bootstrap();
    //单利
    private static RPCRequestNet rpcClient = new RPCRequestNet();
    static { 
        b.group(new NioEventLoopGroup())
        .channel(NioSocketChannel.class)
        .option(ChannelOption.TCP_NODELAY,true);//禁止使用Nagle算法 作用小数据即时传输
    }
    public static RPCRequestNet getRPCRequestNet(){ 
        return rpcClient;
    }
    private RPCRequestNet(){ 
        init();
    }
    public void init() { 
        poolMap = new AbstractChannelPoolMap<InetSocketAddress, FixedChannelPool>() { 
            @Override
            protected FixedChannelPool newPool(InetSocketAddress inetSocketAddress) { 
                ChannelPoolHandler handler = new ChannelPoolHandler() { 
                    //使用完channel需要释放才能放入连接池
                    @Override
                    public void channelReleased(Channel ch) throws Exception { 
                        // TODO Auto-generated method stub
                    }
                    //当链接创建的时候添加channelhandler，只有当channel不足时会创建，但不会超过限制的最大channel数
                    @Override
                    public void channelCreated(Channel ch) throws Exception { 
                        System.out.println("channelCreated. Channel ID: " + ch.id());
                        //ch.pipeline().addLast(new LineBasedFrameDecoder(8192));//以换行符分包
                        //ch.pipeline().addLast(new StringDecoder());//将接收到的对象转为字符串
                        ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535, 0, 2,0,2));
                        ch.pipeline().addLast(new LengthFieldPrepender(2));
                        ch.pipeline().addLast(new ClientMsgPackEncode());//编码
                        ch.pipeline().addLast(new ClientMsgPackDecode());//解码
                        ch.pipeline().addLast(new RPCRequestHandler());//添加相应回调处理
                    }
                    //获取连接池中的channel
                    @Override
                    public void channelAcquired(Channel ch) throws Exception { 
                        // TODO Auto-generated method stub
                    }
                };
                return new FixedChannelPool(b.remoteAddress(inetSocketAddress), handler, 5); //单个服务端连接池大小
            }
        };
    }
    public void connect(String host,int port,final RPCRequest request){ 
        InetSocketAddress addr = new InetSocketAddress(host, port);
        final SimpleChannelPool pool = RPCRequestNet.poolMap.get(addr);
        final Future<Channel> f = pool.acquire();
        f.addListener(new FutureListener<Channel>() { 
            @Override
            public void operationComplete(Future<Channel> arg0) throws Exception { 
                  if (f.isSuccess()) { 
                      Channel ch = f.getNow();
                      ch.writeAndFlush(request);
                      //放回去
                      pool.release(ch);
                  }
            }
        });
        synchronized (request) { 
            //因为异步 所以不阻塞的话 该线程获取不到返回值
            //放弃对象锁 并阻塞等待notify
            try { 
                request.wait();
            } catch (InterruptedException e) { 
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}

Netty客户端处理类

public class RPCRequestHandler extends ChannelInboundHandlerAdapter  {
    public static ChannelHandlerContext channelCtx;
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
         super.channelActive(ctx);
    }
    @Override
    //异步调用读取管道数据
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        RPCResponse response= (RPCResponse) msg;
        //System.out.println("获取到服务器返回值"+responseJson);
        synchronized (RPCProxyHandler.requestLockMap.get(response.getRequestID())) {
            //当客户段获取到返回值的时候唤醒在该对象锁上wait的线程
            RPCRequest request= (RPCRequest) RPCProxyHandler.requestLockMap.get(response.getRequestID());
            //System.out.println("2 : "+response.getResult());
            request.setResult(response.getResult());
            request.notifyAll();
        }
    }
}