好乐买网站推广方式,企业网站开发目的和意义,wordpress管理员登陆,医疗网络营销外包目录 Reactor 模式
具体流程
配置
初始化
NioEventLoop
ServerBootstrapAcceptor 分发 Reactor 模式
在刚学 Netty 的时候#xff0c;我们肯定都很熟悉下面这张图#xff0c;它就是单Reactor多线程模型。 在写Netty 服务端代码的时候#xff0c;下面…目录 Reactor 模式
具体流程
配置
初始化
NioEventLoop
ServerBootstrapAcceptor 分发 Reactor 模式
在刚学 Netty 的时候我们肯定都很熟悉下面这张图它就是单Reactor多线程模型。 在写Netty 服务端代码的时候下面的代码时必不可少的这是为什么呢
public static void main(String[] args) {EventLoopGroup bossGroup new NioEventLoopGroup(1);EventLoopGroup workerGroup new NioEventLoopGroup(4);ServerBootstrap bootstrap new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new NettyServerHandler(), new NettyServerHandler2());System.out.println(netty server start...);bootstrap.bind(9000);
} 在 Netty 里EventLoopGroup 就是线程池不论 bossGroup 还是 workerGroup它们里面的线程都叫 EventLoop。 EventLoopGroup 就是一个线程池bossGroup 叫连接线程池它一般只有一个线程workerGroup 叫工作线程池它一般会有多个线程。bossGroup 线程池里的线程专门监听客户端连接事件监听是否有 SelectionKey.OP_ACCEPT 事件被触发所以 1 个线程就够用了当它监听到有客户端请求连接时它会把这个连接交给 workerGroup 里的一个线程去处理这个过程叫分发这个工作线程会为这个客户端建立一个 NIOSocketChannel并注册到这个工作线程绑定的IO多路复用选择器 Selector 里一个Selector可以接受多个 NIOSocketChannel 的注册所以一个工作线程可以处理多个客户端。 这就是Reactor 模式一个工作线程可以处理多个客户端比 Java 传统的一个客户端对应一个工作线程节约了很多线程减少了大量线程创建线程切换线程销毁的开销所以Netty 性能很好。 上面短短的服务端代码做了很多工作当它刚启动还没有客户端请求连接时bossGroup 连接线程池里的一个线程 EventLoop 会初始化一个 NioServerSocketChannel 并把这个Channel注册到这个EventLoop 持有的IO多路复用选择器Selector里Selector 会监听Channel里的 SelectionKey.OP_ACCEPT 事件一旦有客户端连接过来它会通过下面代码获取到一个
SocketChannel ch javaChannel().accept();
NioSocketChannel并把这个 NioSocketChannel 注册到 workerGroup 工作线程池里的一个EventLoop 里它使用了一个叫 ServerBootstrapAcceptor 的 ChannelInboundHandler接口类去完成这个过程连接完成后后续这个客户端和服务端的交互和数据读写都在这个 EventLoop 完成。
具体流程 下面我们看一下代码Netty 代码中使用了很多继承在继承中可以把子类相同的部分代码提到父类去完成很多子类生成初始化的时候它会调用父类的构造方法去完成这个要注意。
配置 下面的代码主要做一些启动器的配置group(bossGroup, workerGroup) 会设置连接线程池和工作线程池后面有连接事件或读事件过来要处理时它会从这些线程池里取线程去执行channel(NioServerSocketChannel.class) 指定要生成服务端Channel它只会监听SelectionKey.OP_ACCEPT 事件childHandler(new NettyServerHandler(), new NettyServerHandler2()) 是我们业务处理的逻辑。
bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new NettyServerHandler(), new NettyServerHandler2());
初始化 bind() 会把 ServerBootstrapAcceptor 添加到 NioServerSocketChannel 的 pipeline 它会处理连接获取一个 bossGroup 线程池里的 EventLoop并和 NioServerSocketChannel 进行绑定。
bootstrap.bind(9000)
public abstract class AbstractBootstrapB extends AbstractBootstrapB, C, C extends Channel {public void bind(int inetPort) {doBind(new InetSocketAddress(inetPort));} // bind 流程private void doBind(final SocketAddress localAddress) {initAndRegister();// 让channel绑定的线程处理channel.eventLoop().execute(()-{// 绑定指定端口channel.bind(localAddress);});} // 初始化和注册final void initAndRegister() {init(channel);// 把 NioServerSocketChannel 注册到一个复杂连接事件的 EventLoop 的 Selector 里 group.register(channel);} // 把 ServerBootstrapAcceptor 添加到 NioServerSocketChannel 的 pipeline 里abstract void init(Channel channel);
}
NioEventLoop 现在要说一下 NioEventLoop它拥有一个IO多路复用选择器 Selector这个线程会在一个死循环里工作永远也会停止这个线程它会先执行一下 selector.select(1000)阻塞监听1秒看看有没有Channel有事件过来有就去处理任务没有就等待1秒钟再超时放弃再看看自己的任务队列有没有可执行的任务有就去处理任务没有就继续进行死循环继续执行 selector.select(1000)。无论是连接线程还是工作线程都这样处理因为它们共用了这套逻辑。
public class NioEventLoop extends SingleThreadEventLoop {Overrideprotected void run() {for (;;) {try {select();} catch (IOException e) {e.printStackTrace();}try {# 处理事件processSelectedKeys();} finally {runAllTasks();}}}private void select() throws IOException {// 拿到多路复用器Selector selector this.selector;for (;;) {// 等待简化固定1秒int selectedKeys selector.select(1000);// 如果有事件发生或当前有任务跳出循环if (selectedKeys ! 0 || hasTasks()) {break;}}}
}
像下面这种 channel.eventLoop().execute(Runnable), 它也只是把 Runnable 加入到任务处理队列稍后执行。
public abstract class AbstractBootstrapB extends AbstractBootstrapB, C, C extends Channel { private void doBind(final SocketAddress localAddress) {...channel.eventLoop().execute(()-{ channel.bind(localAddress);});}
}
public abstract class SingleThreadEventExecutor implements Executor { // 待执行任务队列private final QueueRunnable taskQueue; Overridepublic void execute(Runnable task) {// 把任务添加到 EventLoop 的任务队列EventLoop 是 SingleThreadEventExecutor 的子类addTask(task);// 执行 EventLoop 的 run 逻辑startThread();}
} 当 NioServerSocketChannel.accept() 监听到一个客户端连接它会把这个 NIOSocketChannel 通过 pipeline 处理最终被 ServerBootstrapAcceptor 所处理
public class NioServerSocketChannel extends AbstractNioMessageChannel {Overrideprotected int doReadMessages(ListObject buf) {SocketChannel ch null;try {ch javaChannel().accept();} catch (IOException e) {}if (ch ! null) {buf.add(new NioSocketChannel(this, ch));return 1;}return 0;}
}
public abstract class AbstractNioMessageChannel extends AbstractNioChannel {Overridepublic void read() {final ChannelPipeline pipeline pipeline();doReadMessages(readBuf);int size readBuf.size();for (int i 0; i size; i ) {pipeline.fireChannelRead(readBuf.get(i));}readBuf.clear();}protected abstract int doReadMessages(ListObject buf);
}
ServerBootstrapAcceptor 分发 ServerBootstrapAcceptor 管理 workerGroup 里的所有工作线程和所有的业务处理代码 ChannelHandlerServerBootstrapAcceptor 会把所有的 ChannelHandler 放到刚刚监听得到的 NIOSocketChannel 里的 pipeline 里并从 workerGroup 里选择一个 EventLoop 工作线程把NIOSocketChannel 注册到该 EventLoop 拥有的IO多路复用选择器 Selector 里去这就完成了分发它已经处理了连接后续这个 NIOSocketChannel 里的所有读写事件都会被 Selector 监听到并被该 EventLoop 工作线程所处理。
private static class ServerBootstrapAcceptor implements ChannelInboundHandler {// 工作线程池即 workerGroup private final EventLoopGroup childGroup;// 业务操作 Handlerprivate final ChannelHandler[] childHandlers;private ServerBootstrapAcceptor(EventLoopGroup childGroup, ChannelHandler[] childHandlers) {this.childGroup childGroup;this.childHandlers childHandlers;}Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {final Channel child (Channel) msg;// 完成 pipeline 责任链模式的组装for (ChannelHandler childHandler : childHandlers) {child.pipeline().addLast(childHandler);}// 把Channel 注册到 SelectorchildGroup.register(child);}Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {// 略}}
