Netty进阶之路--客户端池化

妖狐艹你老母 2022-04-25 02:18 504阅读 0赞

### Netty进阶之路--客户端池化 ###

*  前言
 *  Netty线程池
 *  Netty客户端处理类
 *  效果

# 前言 #

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

# Netty线程池 #

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

废话不多说直接贴代码（把客户端改造了一下 可以看我之前写的 [自己设计一个的轻量级的RPC框架–客户端netty][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();
            }
            
        }
        
    }

# 效果 #

普通调用只需要创建一个足够了，大大的利用了资源。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2EyOTQ2MzQ0NzM_size_16_color_FFFFFF_t_70]  
如果我们通过JMeter 20个用户 调用20次  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2EyOTQ2MzQ0NzM_size_16_color_FFFFFF_t_70 1]

[RPC_netty]: https://blog.csdn.net/a294634473/article/details/88294373
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2EyOTQ2MzQ0NzM_size_16_color_FFFFFF_t_70]: /images/20220210/8d55491d043c4dcba090f4d5fa0b42a0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2EyOTQ2MzQ0NzM_size_16_color_FFFFFF_t_70 1]: /images/20220210/fa6f34ff425c4ee893245ba1533995c3.png