Netty 连接池的使用姿势

摘要: 简单的介绍一下netty连接池的使用

netty版本 4.1.9.Final

一、类介绍

1.ChannelPool ,连接池接口

2.SimpleChannelPool,实现ChannelPool接口，简单的连接池实现

3.FixedChannelPool,继承SimpleChannelPool，有大小限制的连接池实现

4.ChannelPoolMap,管理host与连接池映射的接口

5.AbstractChannelPoolMap,抽象类，实现ChannelPoolMap接口

二、具体使用

我们以http协议的实现来使用连接池，直接上代码

1.ClientPool

public class ClientPool{

//key为目标host，value为目标host的连接池
public ChannelPoolMap<InetSocketAddress, FixedChannelPool> poolMap = null;

public ClientPool(){
init();
}

public void init(){
EventLoopGroup group = new NioEventLoopGroup();
final Bootstrap cb = new Bootstrap();

cb.group(group).channel(NioSocketChannel.class);

poolMap = new AbstractChannelPoolMap<InetSocketAddress, FixedChannelPool>() {
@Override
protected FixedChannelPool newPool(InetSocketAddress key) {
return new FixedChannelPool(cb.remoteAddress(key), new ChannelPoolHandler() {
public void channelReleased(Channel ch) throws Exception {
System.out.println("22");
}

public void channelAcquired(Channel ch) throws Exception {
System.out.println("33");
}

public void channelCreated(Channel ch) throws Exception {
//可以在此绑定channel的handler
ch.pipeline().addLast(new HttpClientCodec())
.addLast(new HttpObjectAggregator(1024 * 1024))
.addLast(new HttpBackendHandler());
}
},20);//单个host连接池大小
}
};

}

public void getHttpClient(InetSocketAddress address,final FullHttpRequest msg) {
if(address == null){
throw new RuntimeException("InetSocketAddress can not be null");
}

final FixedChannelPool pool = this.poolMap.get(address);

Future<Channel> f = pool.acquire();

future.addListener(new FutureListener<Channel>() {

public void operationComplete(Future<Channel> f) {
if (f.isSuccess()) {
Channel ch = f.getNow();
ChannelFuture lastWriteFuture = null;

lastWriteFuture = ch.writeAndFlush(msg);

// Wait until all messages are flushed before closing the channel.
if (lastWriteFuture != null) {

try {
lastWriteFuture.sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}

pool.release(ch);
}
}
});
}

}

2.HttpBackendHandler

public class HttpBackendHandler extends SimpleChannelInboundHandler<FullHttpResponse> {

public HttpBackendHandler() {
}

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("Backend Handler is Active!");
super.channelActive(ctx);
}

@Override
public void channelRead0(final ChannelHandlerContext ctx, FullHttpResponse msg) throws Exception {
//todo something
}

@Override
public void channelInactive(ChannelHandlerContext ctx) {
System.out.println("Backend Handler destroyed!");
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
if (ctx.channel().isActive()) {
ctx.channel().writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
}
}
}

3.使用

public class NettyHttpClientPoolTest {

public static void main(String[] args){
ClientPool pool = new ClientPool();
InetSocketAddress address = new InetSocketAddress("127.0.0.1",8080;
FullHttpRequest request = new DefaultFullHttpRequest(..);
pool.sendMsg(address,request);
}

}

代码有些未完善，需要自己完善一哈，本文主要是做下学习记录