1.netty源码阅读-管理端Server启动
管理端Server
初始化MultiThreadIoEventLoopGroup
1.根据传入的线程数初始化事件执行器
MultithreadEventExecutorGroup构造器
// 初始化时间线程数children=newEventExecutor[nThreads];// 遍历生成for(inti=0;i<nThreads;i++){booleansuccess=false;try{// 根据传入的类型来确认 执行器的类型children[i]=newChild(executor,args);success=true;}catch(Exceptione){...}finally{// 不成功优雅关闭if(!success){...}}}// 所有 EventLoop 创建完成后，再创建选择器。// 当有新任务/新连接到来时，chooser 决定把它分配给哪一个 EventLoop。chooser=chooserFactory.newChooser(children);...readonlyChildren=Collections.unmodifiableSet(childrenSet);MultiThreadIoEventLoopGroup的newChild()方法
protectedIoEventLoopnewChild(Executorexecutor,IoHandlerFactoryioHandlerFactory,@SuppressWarnings("unused")Object...args){returnnewSingleThreadIoEventLoop(this,executor,ioHandlerFactory);}publicSingleThreadIoEventLoop(IoEventLoopGroupparent,Executorexecutor,IoHandlerFactoryioHandlerFactory){super(parent,executor,false,ObjectUtil.checkNotNull(ioHandlerFactory,"ioHandlerFactory").isChangingThreadSupported());this.maxTaskProcessingQuantumNs=DEFAULT_MAX_TASK_PROCESSING_QUANTUM_NS;// io 处理层，EventLoop "怎么去等待 I/O 事件，以及事件来了之后发给谁"。this.ioHandler=ioHandlerFactory.newHandler(this);}看一下SingleThreadIoEventLoop的构造器
/** * parent 线程组归属 * executor 线程的启动方式 * addTaskWakesUp 入队唤醒策略 * supportSuspension 状态机是否含"暂停"状态 * maxPendingTasks 队列容量上限（背压） * rejectedHandler 队列满时的兜底策略 */protectedSingleThreadEventExecutor(EventExecutorGroupparent,Executorexecutor,booleanaddTaskWakesUp,booleansupportSuspension,intmaxPendingTasks,RejectedExecutionHandlerrejectedHandler){super(parent);this.addTaskWakesUp=addTaskWakesUp;this.supportSuspension=supportSuspension;this.maxPendingTasks=Math.max(16,maxPendingTasks);this.executor=ThreadExecutorMap.apply(executor,this);taskQueue=newTaskQueue(this.maxPendingTasks);rejectedExecutionHandler=ObjectUtil.checkNotNull(rejectedHandler,"rejectedHandler");lastActivityTimeNanos=ticker().nanoTime();}初始化ServerBootstrap
ServerBootstrapb=newServerBootstrap();// 简化写法：同一个 EventLoopGroup 同时承担 boss 和 worker 的职责b.group(group).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,100).handler(newLoggingHandler(LogLevel.INFO)).childHandler(newChannelInitializer<SocketChannel>(){@OverridepublicvoidinitChannel(SocketChannelch)throwsException{ChannelPipelinep=ch.pipeline();if(sslCtx!=null){p.addLast(sslCtx.newHandler(ch.alloc()));}//p.addLast(new LoggingHandler(LogLevel.INFO));p.addLast(serverHandler);}});// group 初始化处理线程组。b.group(group) 会把同一个 group 同时赋给 boss 和 worker。// 这是简化用法：一个线程池既接受连接，也处理子连接上的 I/O。// 生产环境中更常见的是显式拆分 bossGroup 和 workerGroup。// EventLoopGroup bossGroup = new NioEventLoopGroup(1); // ① boss 线程池：只接受连接（通常 1~2 个线程）// EventLoopGroup workerGroup = new NioEventLoopGroup(); // ② worker 线程池：处理每个连接上的 I/O（默认 CPU 核数*2）// ServerBootstrap b = new ServerBootstrap();// 显式指定两个线程池：第一个是 boss，第二个是 worker// b.group(bossGroup, workerGroup) // ← parentGroup=bossGroup, childGroup=workerGroup// channel 初始化创建channel的类型，使用工厂设计模式// option 作用于 ServerChannel（NioServerSocketChannel），也就是监听端口、接受连接的那个 Channel。// 整个服务通常只有 1 个 ServerChannel。// handler 作用于 ServerChannel 的 Pipeline，只创建 1 次，贯穿整个服务器生命周期。// childHandler 作用于每个接入的客户端 SocketChannel。// 每个新连接都有自己的 SocketChannel 和 Pipeline，都会执行一次 childHandler 初始化。启动服务
// 同步启动ChannelFuturef=b.bind(PORT).sync();// bind 绑定本地端口号，启动相关任务主要逻辑我们需要看AbstractBootstrap.doBind
privateChannelFuturedoBind(finalSocketAddresslocalAddress){// 初始化服务端 ServerChannel，并将它注册到 parent/boss EventLoopGroupfinalChannelFutureregFuture=initAndRegister();...if(regFuture.isDone()){// 此时我们知道注册已经完成并且成功了。ChannelPromisepromise=channel.newPromise();doBind0(regFuture,channel,localAddress,promise);returnpromise;}else{...// 异常处理returnpromise;}}我们看一下initAndRegister方法
finalChannelFutureinitAndRegister(){Channelchannel=null;try{// 1.创建 Channel 对象（此时还没注册到 EventLoop）channel=channelFactory.newChannel();// 2.调用 init() 方法init(channel);}catch(Throwablet){...// 异常处理}// 3.注册到 EventLoopGroup —— 对 ServerBootstrap 来说，这里注册的是 ServerChannel 到 parent/boss groupfinalChannelFutureregFuture=config().group().register(channel);...returnregFuture;}ServerBootstrap.init方法
voidinit(Channelchannel)throwsThrowable{// 配置 ChannelOption，影响底层 Channel 行为，例如 SO_BACKLOG、SO_REUSEADDR 等setChannelOptions(channel,newOptionsArray(),logger);// 存储用户自定义属性，可以通过 AttributeKey 在 Channel 上保存上下文数据setAttributes(channel,newAttributesArray());// 得到pipeline，pipeline的初始化是在创建 Channel 时完成ChannelPipelinep=channel.pipeline();// 保存子通道配置的引用（避免后续被修改）finalEventLoopGroupcurrentChildGroup=childGroup;finalChannelHandlercurrentChildHandler=childHandler;finalEntry<ChannelOption<?>,Object>[]currentChildOptions=newOptionsArray(childOptions);finalEntry<AttributeKey<?>,Object>[]currentChildAttrs=newAttributesArray(childAttrs);// 加载拓展插件finalCollection<ChannelInitializerExtension>extensions=getInitializerExtensions();// 给 ServerChannel 添加初始化器。这里初始化的是 ServerChannel 的 Pipeline。p.addLast(newChannelInitializer<Channel>(){@OverridepublicvoidinitChannel(finalChannelch){finalChannelPipelinepipeline=ch.pipeline();// 添加用户配置的处理器（如果存在）ChannelHandlerhandler=config.handler();if(handler!=null){pipeline.addLast(handler);}// 这里的 ch 是 ServerChannel，所以任务提交到 boss/parent EventLoop 执行。// 这样可以保证 ServerBootstrapAcceptor 在 handlerAdded 之后再加入 Pipeline。ch.eventLoop().execute(newRunnable(){@Overridepublicvoidrun(){// ServerBootstrapAcceptor 是 ServerChannel Pipeline 中的入站处理器。// 它在接收到 accepted SocketChannel 后：// 1. 给子 Channel 添加 childHandler；// 2. 设置 childOptions 和 childAttrs；// 3. 执行扩展点；// 4. 将子 Channel 注册到 childGroup，也就是 worker EventLoopGroup。pipeline.addLast(newServerBootstrapAcceptor(ch,currentChildGroup,currentChildHandler,currentChildOptions,currentChildAttrs,extensions));}});}});...}关于AbstractChannel的注册流程
publicfinalvoidregister(EventLoopeventLoop,finalChannelPromisepromise){ObjectUtil.checkNotNull(eventLoop,"eventLoop");// 检查...// 设置关联的 EventLoopAbstractChannel.this.eventLoop=eventLoop;// 将通道里面的执行器设置为空AbstractChannelHandlerContextcontext=pipeline.tail;do{context.contextExecutor=null;context=context.prev;}while(context!=null);// 如果当前线程就是 EventLoop 线程，直接注册；否则提交任务到 EventLoop 执行...register0(promise);...}privatevoidregister0(ChannelPromisepromise){...// 创建内部注册 Promise，用于监听底层注册结果ChannelPromiseregisterPromise=newPromise();booleanfirstRegistration=neverRegistered;// 添加监听器，处理注册成功后的回调registerPromise.addListener(future->{if(future.isSuccess()){// 更新注册状态neverRegistered=false;registered=true;// 确保在通知 promise 之前调用 handlerAdded，防止用户在监听器中触发事件时 Handler 还未添加pipeline.invokeHandlerAddedIfNeeded();// 通知 register() 的调用者成功了safeSetSuccess(promise);// 触发注册事件，Pipeline 传播pipeline.fireChannelRegistered();// 仅在首次注册且 Channel 活跃时触发 channelActiveif(isActive()){if(firstRegistration){// 通知我已经准备好了可以进行通信了pipeline.fireChannelActive();}elseif(config().isAutoRead()){// 重新注册且 autoRead 开启时，需要重新开始读取数据beginRead();}}}else{// 注册失败，关闭 Channel 避免资源泄漏...}});// 底层 I/O 注册。以 NIO 为例，这里会把 Channel 注册到 Selector，// 并完成 Channel 与当前 EventLoop 的绑定。doRegister(registerPromise);}