netty版本
- 使用的netty版本是
io.netty:netty-all:4.1.33.Final
Netty耗时任务的处理
- Netty中
EventLoop用来处理IO线程，因此handler中的耗时任务(比如数据库连接、远程调用等)不能在EventLoop里面执行。如果有耗时任务，需要将耗时任务添加到业务线程池中执行。
- 处理耗时任务的两种方式
-
Handler中加入线程池
-
Context中添加线程池
- 两种方式的对比
-
Handler中自定义业务线程池更加灵活，每个Handler都可以自己控制自己的业务线程池，对于非耗时的任务可以不使用业务线程池，直接在EventLoop线程中执行。
-
Context中添加线程池(Netty建议的方式),在pipeline中添加Handler的时候，添加一个业务线程池。这种方式Handler中的代码不需要做任何修改，但是整个Handler都交给业务线程池，无论是否是耗时任务都会加入到队列里，控制的粒度比较粗。
- 如果业务
ChannelHandler处理逻辑比较简单，执行时间是受控的，业务I/O线程的负载也不重，在这种应用场景下，业务ChannelHandler可以和I/O操作共享同一个线程。使用这种线程模型会带来两个优势：
- 开发简单，开发业务
ChannelHandler的不需要关注Netty的线程模型
- 性能更高：因为减少了一次线程上下文切换，所以性能会更高。
- Netty线程池的特点
-
Channel和EventLoop是N:1的关系：一个Channel生命周期内只注册一个EventLoop，一个EventLoop可能会给分配给多个Channel,所以如果一个Channel阻塞可能会导致其他在同一个EventLoop上的Channel都阻塞。
- 一个
EventLoopGroup包含一个或者多个EventLoop,一个EventLoop在其生命周期内只与一个Thread绑定（EbeddedEventLoop除外）EventLoop的Thread负责处理该事件循环内所有的IO事件
代码案例
Handler自定义业务线程池
-
Handler中自定义业务线程池
@ChannelHandler.Sharable
public class EchoServerHandler extends SimpleChannelInboundHandler<String> {
public static final ChannelHandler INSTANCE = new EchoServerHandler();
private static final String LINE = System.getProperty("line.separator");
private EchoServerHandler() {
}
protected static ExecutorService newFixedThreadPool() {
final ThreadFactory threadFactory = new ThreadFactoryBuilder()
.setNameFormat("netty-business-%d")
.setDaemon(false)
.build();
return new ThreadPoolExecutor(200, 200,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(10000),
threadFactory);
}
final static ListeningExecutorService service = MoreExecutors.listeningDecorator(newFixedThreadPool());
@Override
public void channelRead0(ChannelHandlerContext ctx, String msg) {
service.submit(new Runnable() {
@Override
public void run() {
//模拟耗时任务
try {
Thread.sleep(5000);
logger.info("execute time 5s");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
ctx.writeAndFlush(msg + LINE).addListener(ChannelFutureListener.CLOSE);
}
});
}
}
-
Server启动代码
public class EchoServer {
private final int port;
public EchoServer(int port) {
this.port = port;
}
public void start() throws Exception {
NioEventLoopGroup boss = new NioEventLoopGroup(1);
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(boss, worker)
.channel(NioServerSocketChannel.class)
.localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new LineBasedFrameDecoder(1024));
pipeline.addLast(new StringDecoder());
pipeline.addLast(new StringEncoder());
pipeline.addLast(EchoServerHandler.INSTANCE);
}
});
ChannelFuture f = b.bind().sync();
System.out.println(String.format("%s started and listen on %s", EchoServer.class.getName(), f.channel().localAddress()));
f.channel().closeFuture().sync();
} finally {
boss.shutdownGracefully().sync();
worker.shutdownGracefully().sync();
}
}
public static void main(String[] args) throws Exception {
new EchoServer(8080).start();
}
}
-
使用多个telnet连接并请求，输出结果,可以看出执行任务使用的是我们自己定义的线程池
netty-business-0 execute time 5s
netty-business-1 execute time 5s
Context中添加线程池
-
Handler代码无需任何的改动,不需要启动任何额外线程来处理任务
@ChannelHandler.Sharable
public class EchoServerHandler extends SimpleChannelInboundHandler<String> {
public static final ChannelHandler INSTANCE = new EchoServerHandler();
private static final String LINE = System.getProperty("line.separator");
private EchoServerHandler() {
}
@Override
public void channelRead0(ChannelHandlerContext ctx, String msg) {
//模拟耗时任务
try {
Thread.sleep(5000);
System.out.printf("%s execute time 5s \n", Thread.currentThread().getName());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
ctx.writeAndFlush(msg + LINE).addListener(ChannelFutureListener.CLOSE);
}
}
-
启动代码
public class EchoServer {
private final int port;
public EchoServer(int port) {
this.port = port;
}
public void start() throws Exception {
final ThreadFactory threadFactory = new ThreadFactoryBuilder()
.setNameFormat("netty-context-business-%d")
.setDaemon(false)
.build();
NioEventLoopGroup boss = new NioEventLoopGroup(1);
NioEventLoopGroup worker = new NioEventLoopGroup();
NioEventLoopGroup business = new NioEventLoopGroup(200,threadFactory);
try {
ServerBootstrap b = new ServerBootstrap();
b.group(boss, worker)
.channel(NioServerSocketChannel.class)
.localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new LineBasedFrameDecoder(1024));
pipeline.addLast(new StringDecoder());
pipeline.addLast(new StringEncoder());
pipeline.addLast(business, EchoServerHandler.INSTANCE);
}
});
ChannelFuture f = b.bind().sync();
System.out.println(String.format("%s started and listen on %s", EchoServer.class.getName(), f.channel().localAddress()));
f.channel().closeFuture().sync();
} finally {
boss.shutdownGracefully().sync();
worker.shutdownGracefully().sync();
business.shutdownGracefully().sync();
}
}
public static void main(String[] args) throws Exception {
new EchoServer(8080).start();
}
}
-
使用多个telnet连接并请求，输出结果
netty-context-business-0 execute time 5s
netty-context-business-1 execute time 5s
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
