1、Netty工作原理示意图1
Netty主要基于主从Reactors多线程模型做了一定改进,具有多个Reactor;
- BossGroup线程维护Selector,只关注Accept;
- 当接收到Accept事件,获取到对应的SocketChannel,封装成NIOSocketChannel并注册到Worker线程,并进行维护;
- 当worker线程监听到selector中通道发生自己感兴趣的事件后,就由handler进行处理;
2、Netty工作原理示意图2
3、Netty工作原理图3
- Netty抽象出两组线程池:BossGroup专门负责接收客户端的连接,WorkerGroup专门负责网络的读写;他们都是NioEventLoopGroup;
- NioEventLoopGroup相当于一个事件循环组,这个组中含有多个事件循环,每一个都是NioEventLoop;
- NioEventLoop表示一个不断循环的执行处理任务的线程,每一个NioEventLoop都有一个selector,用于监听绑定在其上的socket网络通讯;
- NioEventLoopGroup可以有多个线程,即含有多个NioEventLoop;
- 每个Boss NioEventLoop循环执行的步骤包括三步:
轮询accept事件;
处理accept事件,与client建立连接,生成NioSocketChannel,并将其注册到某个Worker NioEventLoop的selector;
处理任务队列的任务,即runAllTasks
- Worker NioEventLoop循环执行的步骤:
轮询read,write事件
处理io事件,即read/write事件,在对应的NioSocketChannel处理;
处理任务队列的任务,即runAllTasks;
- 每个worker NioEventLoop处理业务时,会使用pipeline(管道),里面包含了channel,即通过pipeline可以获取到对应通道,管道中维护了很多的handler;
4、Netty实例-TCP服务
package gyx.netty.simple;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class NettyServer {
public static void main(String[] args) throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try{
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup,workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG,128)
.childOption(ChannelOption.SO_KEEPALIVE,true)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new NettyServerHandler());
}
});
System.out.println(".....服务器 is ready...");
ChannelFuture cf = bootstrap.bind(6668).sync();
cf.channel().closeFuture().sync();
}finally{
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
package gyx.netty.simple;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("服务器读取线程 "+Thread.currentThread().getName());
System.out.println("server ctx="+ctx);
Channel channel = ctx.channel();
ByteBuf buf=(ByteBuf)msg;
System.out.println("客户端发送消息是:"+buf.toString(CharsetUtil.UTF_8));
System.out.println("客户端地址:"+channel.remoteAddress());
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(Unpooled.copiedBuffer("hello,client",CharsetUtil.UTF_8));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
package gyx.netty.simple;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
public class NettyClient {
public static void main(String[] args) throws InterruptedException {
EventLoopGroup group = new NioEventLoopGroup();
try{
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new NettyClientHandler());
}
});
System.out.println("client ok..");
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();
channelFuture.channel().closeFuture().sync();
}finally {
group.shutdownGracefully();
}
}
}
package gyx.netty.simple;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
public class NettyClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("client "+ctx);
ctx.writeAndFlush(Unpooled.copiedBuffer("hello,server", CharsetUtil.UTF_8));
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf=(ByteBuf)msg;
System.out.println("服务器回复的消息:"+buf.toString(CharsetUtil.UTF_8));
System.out.println("服务器地址:"+ctx.channel().remoteAddress());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}