跳至主要内容

Netty入门

橘子coding2026/10/9约 11305 字

1. Reactor 模式

Reactor 模式是网络编程里最经典的事件驱动设计模式，Netty、Redis、Nginx、Tomcat 里都能看到它。它的核心思想可以用三步概括：

注册感兴趣的事件 → 扫描是否有感兴趣的事件发生 → 事件发生后做出相应的处理。

1.1 核心思想

对照 NioSelectorServer 的代码，这三步的对应关系非常直接：

Reactor 三步NIO 里的实现
注册感兴趣的事件channel.register(selector, SelectionKey.OP_READ)
扫描是否有事件发生selector.select()，阻塞等待，由内核通知就绪事件
事件发生后处理遍历 selectedKeys()，按事件类型分发到对应的处理逻辑

其中 Selector 负责「扫描」，register 负责「注册」，「分发」由应用代码完成。把这套流程抽象出来就是 Reactor 模式。接下来用 Doug Lea《Scalable IO in Java》里的四张图，看这套模式怎么从最朴素的形态一步步演进到主从 Reactor；先说明一下图里反复出现的那些名词。

1.2 图里的名词说明

四张图中间都横着一排 read → decode → compute → encode → send，这是一条连接处理一个请求的完整流程。四张图的差别，本质就是这五步分别放在哪个线程里执行。

名词作用属于
client客户端，发起连接请求的一方；图上画多个，代表并发连接的规模角色
Server服务端，监听端口、接受连接的一方角色
Reactor事件循环本体，内部持有一个 Selector，只负责「等事件」和「分事件」角色
dispatch事件分发。Reactor 拿到就绪事件后，按事件类型（accept、read、write）投给对应的处理逻辑角色
acceptor专门处理 accept 的角色，负责接下新连接并把它注册到 Reactor；在 Netty 里对应 ServerBootstrap$ServerBootstrapAcceptor，挂在 boss 线程上角色
handler一条连接从 read 到 send 的完整处理链；这也是 Netty 里 ChannelHandler + ChannelPipeline 的原型角色
Thread Pool业务线程池。decode / compute / encode 太重、不适合放在 Reactor 线程里做时，就丢给它执行角色
worker threads线程池里的工作线程，真正跑 decode / compute / encode 的线程角色
queued tasks任务队列。worker 线程都忙的时候，新提交的任务先在这里排队等着角色
mainReactor主 Reactor，只负责 accept角色
subReactor从 Reactor，负责已建立连接的 read / send；和 mainReactor 各自独立，互不拖累角色
read从 socket 读数据，把内核接收缓冲区里的字节拷进用户态；对应上一篇「IO模型」里 NIO 的 socketChannel.read(byteBuffer)IO 步骤
send把数据写回 socket；对应 socketChannel.write(byteBuffer)IO 步骤
decode解码。网络上流动的是字节，业务要的是消息对象，这一步按协议把字节还原成消息，顺带处理粘包半包业务步骤
compute业务计算。真正干活的逻辑，比如查库、算价、拼响应，通常是整条链上最耗时的一步业务步骤
encode编码，decode 的逆操作，把业务对象按协议序列化成字节业务步骤

read / send 是 IO 步骤，必须贴着 socket 做，天然属于持有 Selector 的那个线程；decode / compute / encode 是业务步骤，可以搬到别的线程去。后面四张图看的，就是这三步被搬到了哪里，以及为什么值得搬。

1.3 经典服务设计：一连接一线程

Netty入门-Classic Service Designs

Server 每接受一个连接就分配一个 handler 线程，线程里 read → decode → compute → encode → send 串行走完，一个请求处理完再处理下一个。这就是上一篇「IO模型」里 BIO 一连接一线程的模型图。

问题在于线程数跟连接数绑死了：连接闲着的时候，处理它的线程只能跟着等；再加上线程栈和上下文切换的开销，连接一多，服务端就撑不住。所以接下来的问题就变成了——怎么用少量线程盯住大量连接。

1.4 单 Reactor 单线程

Netty入门-Basic Reactor Design

一个 Reactor 线程用 Selector 监听所有连接：acceptor 只负责 accept，把新连接注册到 Reactor；连接上有事件就绪时，dispatch 把它分发给对应的 read / send；read 之后的 decode / compute / encode 也在同一个 Reactor 线程里做完，最后 send 回去。

好处是整个服务只有一个线程在跑，没有多线程竞争，实现最简单。Redis 就是这种模型，Redis 6 之后网络 IO 引入了多线程，但核心命令处理仍然是单线程。

缺点是 decode / compute / encode 属于业务步骤，一旦耗时就会卡住事件循环——这期间 accept 和其他连接的读写都只能排队等着。

1.5 单 Reactor 多线程

Netty入门-Worker Thread Pools

既然卡住的根源是「业务步骤和 IO 步骤挤在同一个线程里」，那就把它们分开：把 decode / compute / encode 从 Reactor 线程挪进 Thread Pool，交给 worker threads 执行；线程池忙不过来时，任务先放进 queued tasks 排队。Reactor 线程只保留 read / send，于是它能一直快速地转下去，新到的事件不会因为某个业务慢而被积压。

代价是引入了新的复杂度：线程池大小、队列积压和拒绝策略都要考虑；另外同一个连接的多个处理任务如果并行执行，返回顺序可能会乱，需要按连接保证顺序。

1.6 主从 Reactor 多线程

Netty入门-Using Multiple Reactors

上一个模型里，accept 和成千上万条连接的读写仍然挤在同一个 Reactor 线程上，建连高峰会拖慢已有连接的读写。于是再拆一层：mainReactor 只负责 accept，拿到连接后注册到 subReactor；subReactor 负责 read / send 的分发；decode / compute / encode 依旧交给 worker threads 和 queued tasks；这样连接建立和读写就能各自独立扩展，互不拖累。

对应到 Netty：bossGroup 就是 mainReactor，workerGroup 里的一组事件循环就是 subReactor（4.1.x 叫 NioEventLoop，4.2.x 叫 SingleThreadIoEventLoop）。另外 Netty 默认连业务处理也在 worker 线程里执行，需要把业务隔离出去时，才额外挂一个业务线程池。

1.7 四种模型对比

模型连接建立 accept读写事件 read / send业务处理 decode / compute / encode典型代表
经典服务设计每连接一个线程同一个线程同一个线程BIO
单 Reactor 单线程Reactor 线程Reactor 线程Reactor 线程Redis（6 之前）
单 Reactor 多线程Reactor 线程Reactor 线程线程池—
主从 Reactor 多线程mainReactorsubReactor线程池（Netty 默认仍在 worker 线程）Netty

一句话总结 Reactor 的价值：它把「等待事件」和「处理事件」解耦，等待交给少量线程（甚至一个），处理可以按需扩展，从而用有限的线程支撑海量连接——这正是 Netty 高性能的根基。

2. Netty 使用场景

直接用 JDK 自带的 NIO 写网络程序，门槛并不低。一方面是 API 繁杂，Selector、ServerSocketChannel、SocketChannel、ByteBuffer 这一堆类得先熟悉；另一方面真正的麻烦不在 API 上，而在网络环境本身的不可靠：断线重连、网络闪断、心跳、半包读写、网络拥塞、异常流处理，NIO 一概不管，全得业务代码自己兜。

Netty 要解决的就是这两件事。它在 JDK NIO 之上做了一层封装，把那些通用又琐碎的网络处理收进框架里，业务代码只需要关心自己的逻辑；同时性能上也有优势，吞吐量更高、延迟更低、资源消耗更少，并且尽量减少不必要的数据复制。

版本方面，现在通用的是 4.x，5.x 已经废弃。4.2 的编译基线是 Java 8，本示例跑在 Java 21 上。

使用场景主要有三类：

  1. 互联网行业：分布式系统里各节点之间要远程调用，高性能的 RPC 框架是刚需，Netty 作为异步高性能的通信组件，常常被这些 RPC 框架当作底层通信设施。最典型的是 Dubbo，Dubbo 协议默认使用 Netty 作为基础通信组件，实现各进程节点之间的内部通信；RocketMQ 底层用的也是 Netty。Zookeeper、gRPC-Java、Spring WebFlux 的底层同样站着 Netty。
  2. 游戏行业：不管是手游服务端还是大型网络游戏，Java 的占比都不低。Netty 本身提供了 TCP/UDP 和 HTTP 协议栈，可以直接作为高性能基础通信组件使用。
  3. 大数据领域：Hadoop 的高性能通信和序列化组件 Avro，它的 RPC 框架默认采用 Netty 进行跨节点通信，Avro 的 Netty Service 也是在 Netty 之上二次封装的。

更多基于 Netty 的开源项目见 Netty related projects。

3. Netty 入门示例

先引入 Maven 依赖，示例使用 Netty 4.2.18.Final：

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.2.18.Final</version>
</dependency>

Netty 4.2 里 NioEventLoopGroup 已经过时

4.2 之前创建线程组的写法是 new NioEventLoopGroup()，4.2 开始 NioEventLoopGroup 和 NioEventLoop 都被标记为 @Deprecated，取而代之的是 MultiThreadIoEventLoopGroup 配合 NioIoHandler.newFactory()。这个改动把「事件循环」和「IO 实现」解耦了：线程组只负责调度和管理线程，具体用 NIO 还是 epoll，由注入的 IoHandlerFactory 决定。

3.1 服务端

package cn.juzicoding.netty.base;

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.MultiThreadIoEventLoopGroup;
import io.netty.channel.nio.NioIoHandler;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * Netty 服务端示例
 *
 * @author 橘子coding
 */
public class NettyServer {

    public static void main(String[] args) throws Exception {

        // 创建两个线程组 bossGroup 和 workerGroup，含有的子线程 NioEventLoop 的个数默认为 CPU 核数的两倍
        // bossGroup 只是处理连接请求，真正的和客户端业务处理，会交给 workerGroup 完成
        EventLoopGroup bossGroup = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory());
        EventLoopGroup workerGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
        try {
            // 创建服务器端的启动对象
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 使用链式编程来配置参数
            // 设置两个线程组
            bootstrap.group(bossGroup, workerGroup)
                    // 使用 NioServerSocketChannel 作为服务器的通道实现
                    .channel(NioServerSocketChannel.class)
                    // 初始化服务器连接队列大小，服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接
                    // 多个客户端同时来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    // 创建通道初始化对象，设置初始化参数
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // 对 workerGroup 的 SocketChannel 设置处理器
                            ch.pipeline().addLast(new NettyServerHandler());
                        }
                    });
            System.out.println("netty server start。。");
            // 绑定一个端口并且同步，生成了一个 ChannelFuture 异步对象，通过 isDone() 等方法可以判断异步事件的执行情况
            // 启动服务器（并绑定端口），bind 是异步操作，sync 方法是等待异步操作执行完毕
            ChannelFuture cf = bootstrap.bind(9000).sync();
            // 给 cf 注册监听器，监听我们关心的事件
            /* cf.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (cf.isSuccess()) {
                        System.out.println("监听端口9000成功");
                    } else {
                        System.out.println("监听端口9000失败");
                    }
                }
            }); */
            // 对通道关闭进行监听，closeFuture 是异步操作，监听通道关闭
            // 通过 sync 方法同步等待通道关闭处理完毕，这里会阻塞等待通道关闭完成
            cf.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

几处值得留意的地方：

  • MultiThreadIoEventLoopGroup(1, ...) 里的 1 表示 bossGroup 只用一个线程，它只负责接收连接；workerGroup 不指定线程数，默认是 CPU 核数的 2 倍，真正的读写和业务处理都交给它。
  • .option(ChannelOption.SO_BACKLOG, 1024) 设置的是服务端连接队列的大小。服务端处理连接请求是顺序的，同一时间只能处理一个，来不及处理的连接请求会先放进队列等待。
  • .childHandler(...) 作用于 workerGroup 上的 SocketChannel；.handler(...) 作用于 bossGroup 的 ServerSocketChannel，这里没有用到。
  • bind() 是异步的，sync() 只是在这里等绑定完成；后面的 closeFuture().sync() 会阻塞主线程，直到通道关闭。

自定义服务端 Handler：

package cn.juzicoding.netty.base;

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;

/**
 * 自定义 Handler 需要继承 netty 规定好的某个 HandlerAdapter（规范）
 *
 * @author 橘子coding
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 读取客户端发送的数据
     *
     * @param ctx 上下文对象，含有通道 channel，管道 pipeline
     * @param msg 就是客户端发送的数据
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("服务器读取线程 " + Thread.currentThread().getName());
        // Channel channel = ctx.channel();
        // 本质是一个双向链接，出站入站
        // ChannelPipeline pipeline = ctx.pipeline();
        // 将 msg 转成一个 ByteBuf，类似 NIO 的 ByteBuffer
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("客户端发送消息是:" + buf.toString(CharsetUtil.UTF_8));
    }

    /**
     * 数据读取完毕处理方法
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ByteBuf buf = Unpooled.copiedBuffer("HelloClient", CharsetUtil.UTF_8);
        ctx.writeAndFlush(buf);
    }

    /**
     * 处理异常，一般是需要关闭通道
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

channelRead 里的 msg 就是客户端发来的数据，这里直接强转成 ByteBuf；channelReadComplete 在一次读取完成后触发，用它把 HelloClient 回写给客户端；exceptionCaught 处理异常，一般是直接关掉通道。

3.2 客户端

package cn.juzicoding.netty.base;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.MultiThreadIoEventLoopGroup;
import io.netty.channel.nio.NioIoHandler;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * Netty 客户端示例
 *
 * @author 橘子coding
 */
public class NettyClient {
    public static void main(String[] args) throws Exception {
        // 客户端需要一个事件循环组
        EventLoopGroup group = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
        try {
            // 创建客户端启动对象
            // 注意客户端使用的不是 ServerBootstrap 而是 Bootstrap
            Bootstrap bootstrap = new Bootstrap();
            // 设置相关参数
            // 设置线程组
            bootstrap.group(group)
                    // 使用 NioSocketChannel 作为客户端的通道实现
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            // 加入处理器
                            channel.pipeline().addLast(new NettyClientHandler());
                        }
                    });
            System.out.println("netty client start");
            // 启动客户端去连接服务器端
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9000).sync();
            // 对关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

客户端用的是 Bootstrap 而不是 ServerBootstrap，通道实现也换成了 NioSocketChannel。

自定义客户端 Handler：

package cn.juzicoding.netty.base;

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;

/**
 * 自定义客户端 Handler
 *
 * @author 橘子coding
 */
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    /**
     * 当客户端连接服务器完成就会触发该方法
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf buf = Unpooled.copiedBuffer("HelloServer", CharsetUtil.UTF_8);
        ctx.writeAndFlush(buf);
    }

    // 当通道有读取事件时会触发，即服务端发送数据给客户端
    @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();
    }
}

channelActive 在连接建立完成后触发，用它发出 HelloServer；channelRead 处理服务端返回的数据。

先启动 NettyServer，再启动 NettyClient，服务端输出：

netty server start。。
服务器读取线程 multiThreadIoEventLoopGroup-3-1
客户端发送消息是:HelloServer

客户端输出：

netty client start
收到服务端的消息:HelloClient
服务端的地址： /127.0.0.1:9000

这里的 multiThreadIoEventLoopGroup-3-1 就是 workerGroup 里干活的线程，也说明 read 事件确实交给了 Worker 处理，而不是 Boss。另外要注意，客户端的 closeFuture().sync() 会一直等下去，示例里的连接不会主动关闭，所以跑完客户端进程不会退出，需要手动停。

看完这段代码会发现，Netty 的目标就是把业务逻辑从网络基础处理里剥出来：连接的建立、事件的监听与分发、读写的调度都由框架完成，业务只需要在 Handler 里关心「收到什么、回什么」，不用再写一大堆类似 NIO 的网络处理代码。

4. Netty 线程模型

Netty 的线程模型建立在《Scalable IO in Java》里那套 IO 处理模式之上。从 4.1.x 到 4.2.x，BossGroup / WorkerGroup 的划分、事件循环的循环步骤、pipeline 的走向都没有变，变的是事件循环这一层的类名和职责划分，两张图对照着看最清楚。

4.1 Netty 4.1.x 的 Reactor 结构

Netty入门-Netty4.1.x Reactor工作架构图

对照这张图：

  1. Netty 抽象出两组线程池 BossGroup 和 WorkerGroup：BossGroup 专门负责接收客户端的连接，WorkerGroup 专门负责网络的读写；
  2. BossGroup 和 WorkerGroup 的类型都是 NioEventLoopGroup；
  3. NioEventLoopGroup 相当于一个事件循环线程组，组里含有多个事件循环线程，每一个事件循环线程就是一个 NioEventLoop；
  4. 每个 NioEventLoop 都有一个 Selector，用于监听注册在它上面的 SocketChannel 的网络通信；
  5. Boss 的 NioEventLoop 线程内部循环执行 3 步：select（等待事件就绪）→ processSelectedKeys（处理就绪事件，这里就是 accept，拿到新连接后把 channel 注册到 Worker 的 Selector）→ runAllTasks（执行任务队列）；
  6. Worker 的 NioEventLoop 线程循环执行的步骤同样是这三步，区别在于 processSelectedKeys 里处理的是 read / write 事件；
  7. Worker 的 NioEventLoop 处理 NioSocketChannel 的业务时，会使用 pipeline（管道），管道中维护了很多 handler 处理器，用来处理 channel 中的数据。

这一版里事件循环和 NIO 是绑死的：NioEventLoop 自己持有 Selector，自己 select、自己处理就绪的 SelectionKey，连类名里都带着 Nio。

4.2 Netty 4.2.x 的 Reactor 结构

Netty入门-Netty4.2.x Reactor工作架构图

4.2.x 的整体结构和 4.1.x 一模一样，只是每个事件循环从 NioEventLoop 换成了 SingleThreadIoEventLoop——类名里的 Nio 没了。这不是单纯改名，而是把「事件循环」和「IO 实现」拆成了两层：

4.1.x4.2.x变化
NioEventLoopGroupMultiThreadIoEventLoopGroup线程组不再和 NIO 绑定，创建时要传入 IoHandlerFactory，用 NIO 就是 NioIoHandler.newFactory()
NioEventLoopSingleThreadIoEventLoop事件循环本身不再认识 NIO，也不再直接持有 Selector
NioEventLoop 里的 SelectorNioIoHandler 里的 Selectorselect、处理就绪事件都交给 IoHandler 做
SelectionKeyIoRegistration注册的返回值换成 IoRegistration，通过 submit(IoOps) 表达关心的事件

对照图上的三步，落到 4.2 的代码里其实变了样：select 和 processSelectedKeys 都发生在 NioIoHandler.run(IoHandlerContext) 内部，SingleThreadIoEventLoop 只负责跑线程、跑任务队列、调用 IoHandler：

// io.netty.channel.SingleThreadIoEventLoop#run（4.2.18，省略断言）
protected void run() {
    ioHandler.initialize();
    do {
        // 内部就是 ioHandler.run(context)，4.2 里 select 和 processSelectedKeys 都在里面
        runIo();
        if (isShuttingDown()) {
            ioHandler.prepareToDestroy();
        }
        runAllTasks(maxTaskProcessingQuantumNs);
    } while (!confirmShutdown() && !canSuspend());
}

所以图上把 Selector 画在事件循环里，从「这个 Selector 服务于哪个事件循环」的角度看没问题；只是 4.2 的实现里它归 NioIoHandler 持有，事件循环通过 ioHandler() 拿到它。

NioEventLoop 和 NioEventLoopGroup 在 4.2 里并没有删掉，而是变成了 @Deprecated 的兼容类（NioEventLoop extends SingleThreadIoEventLoop、NioEventLoopGroup extends MultiThreadIoEventLoopGroup），老代码照样跑，新代码应该换成新 API。拆开的好处是同一套 group / loop 可以配不同的 IO 实现：Linux 上想换成 epoll，只要把工厂换成 EpollIoHandler.newFactory()，事件循环这一层不用动。

这也正好对上第 1 节 Reactor 模式的划分：BossGroup 就是主 Reactor，只做 accept；WorkerGroup 是从 Reactor，负责读写事件的分发；pipeline 上的业务处理最终落在 Worker 的线程上。

5. Netty 核心组件

Bootstrap / ServerBootstrap

引导类。一个 Netty 应用通常由一个 Bootstrap 开始，主要作用是配置整个 Netty 程序、把各个组件串联起来。Netty 中 Bootstrap 是客户端程序的启动引导类，ServerBootstrap 是服务端启动引导类。

Future / ChannelFuture

Netty 中所有的 IO 操作都是异步的，调用之后不能立刻得知消息是否被正确处理，所以这些调用都会返回一个 Future。可以过一会再等它执行完成，或者直接注册一个监听器（addListener()），操作成功或失败时监听器会自动触发。入门示例里的 bootstrap.bind(9000).sync() 和 closeFuture().sync() 用的就是这个机制。

Channel

Netty 网络通信的组件，用于执行网络 IO 操作。Channel 为用户提供：

  1. 当前网络连接的通道的状态（例如是否打开、是否已连接）；
  2. 网络连接的配置参数（例如接收缓冲区大小）；
  3. 异步的网络 IO 操作（如建立连接、读写、绑定端口），异步意味着任何 IO 调用都将立即返回，并且不保证在调用结束时所请求的 IO 操作已经完成；
  4. 调用立即返回一个 ChannelFuture 实例，通过往 ChannelFuture 上注册监听器，可以在 IO 操作成功、失败或取消时得到回调通知；
  5. 支持关联 IO 操作与对应的处理程序。

不同协议、不同阻塞类型的连接都有不同的 Channel 类型与之对应，常用的有：

Channel说明
NioSocketChannel异步的客户端 TCP Socket 连接
NioServerSocketChannel异步的服务器端 TCP Socket 连接
NioDatagramChannel异步的 UDP 连接
NioSctpChannel异步的客户端 SCTP 连接
NioSctpServerChannel异步的服务器端 SCTP 连接

这些通道涵盖了 UDP 和 TCP 网络 IO 以及文件 IO。这些通道在这一层没有受 4.2 的影响，类名照旧。

Selector

Netty 基于 Selector 对象实现 IO 多路复用，通过 Selector，一个线程可以监听多个连接的 Channel 事件。向 Selector 注册 Channel 之后，Selector 内部的机制会自动不断地查询这些注册的 Channel 是否有已就绪的 IO 事件（例如可读、可写、网络连接完成等），这样一个线程就能高效地管理多个 Channel。

4.2 开始 Selector 不再由事件循环直接持有，而是交给 NioIoHandler，事件循环通过 IoHandler 使用它，见下一条。

IoHandler / IoHandlerFactory

这两个是 4.2 新抽出来的一层，用来把「事件循环」和「IO 实现」解耦：

  • IoHandler：真正的 IO 实现，负责 select、处理就绪事件、注册 IoHandle，以及唤醒和销毁。NIO 对应 NioIoHandler（Selector 就在它内部），Linux 上的 epoll 对应 EpollIoHandler；
  • IoHandlerFactory：创建 IoHandler 的工厂。线程组创建时把工厂传进去，每个事件循环启动时用它 new 一个自己的 IoHandler。
// 创建线程组时注入用哪种 IO 实现
EventLoopGroup bossGroup = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory());
EventLoopGroup workerGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());

换成 epoll 只需要把 NioIoHandler.newFactory() 改成 EpollIoHandler.newFactory()。

SingleThreadIoEventLoop

4.1.x 里这个角色叫 NioEventLoop，4.2 改名为 SingleThreadIoEventLoop：内部维护了一个线程、一个任务队列和一个 IoHandler，支持异步提交执行任务。线程启动时会调用 run 方法，循环处理两类任务：

  • IO 处理：runIo()，内部就是 ioHandler.run(context)，也就是 4.1.x 里的 select + processSelectedKeys，就绪事件 accept、connect、read、write 都在这一步处理；
  • 非 IO 任务：runAllTasks()，处理添加到 taskQueue 中的任务，如 register0、bind0。

MultiThreadIoEventLoopGroup

4.1.x 里这个角色叫 NioEventLoopGroup，4.2 改名为 MultiThreadIoEventLoopGroup：主要管理 eventLoop 的生命周期，可以理解为一个线程池，内部维护了一组线程，每个线程（SingleThreadIoEventLoop）负责处理多个 Channel 上的事件，而一个 Channel 只对应于一个线程。

Channel 和线程是绑定的

一个 Channel 上的事件永远由同一个线程处理，所以 Handler 里不用为共享状态加锁；反过来，Handler 里只要有一个阻塞操作，就会卡住这个线程上其他所有 Channel 的事件。

ChannelHandler

ChannelHandler 是一个接口，处理 IO 事件或拦截 IO 操作，并将其转发到其 ChannelPipeline（业务处理链）中的下一个处理程序。

这个接口本身有很多方法需要实现，直接实现不方便，所以通常继承它的子类：

ChannelInboundHandler    用于处理入站 IO 事件
ChannelOutboundHandler   用于处理出站 IO 操作

或者使用适配器类：

ChannelInboundHandlerAdapter    用于处理入站 IO 事件
ChannelOutboundHandlerAdapter   用于处理出站 IO 操作

ChannelHandlerContext

保存 Channel 相关的所有上下文信息，同时关联一个 ChannelHandler 对象。

ChannelPipeline

ChannelPipeline 保存 ChannelHandler 的 List，用于处理或拦截 Channel 的入站事件和出站操作。

ChannelPipeline 实现了一种高级形式的拦截过滤器模式，使用户可以完全控制事件的处理方式，以及各个 ChannelHandler 之间如何相互交互。在 Netty 中，每个 Channel 都有且仅有一个 ChannelPipeline 与之对应：

Netty入门-ChannelPipeline 结构

一个 Channel 包含一个 ChannelPipeline，而 ChannelPipeline 中又维护了一个由 ChannelHandlerContext 组成的双向链表，并且每个 ChannelHandlerContext 中又关联着一个 ChannelHandler。

read 事件（入站事件）和 write 事件（出站事件）走在同一个双向链表上：入站事件会从链表 head 往后传递到最后一个入站的 handler，出站事件会从链表 tail 往前传递到最前一个出站的 handler，两种类型的 handler 互不干扰。

6. ByteBuf

从结构上来说，ByteBuf 由一串字节数组构成，数组中每个字节用来存放信息。

ByteBuf 提供了两个索引，一个用于读取数据，一个用于写入数据，这两个索引通过在字节数组中移动，来定位需要读或者写信息的位置。当从 ByteBuf 读取时，它的 readerIndex（读索引）会根据读取的字节数递增；同样，当写 ByteBuf 时，它的 writerIndex 也会根据写入的字节数递增。

Netty入门-ByteBuf 读写索引

以 Unpooled.buffer(10) 为例，capacity 是 10，两个索引把整个数组切成三个区域：

区域范围说明
已读区域[0, readerIndex)已经读过的字节，想重读需要手动把 readerIndex 移回去
可读区域[readerIndex, writerIndex)readableBytes() 的大小，也是 readXxx() 真正能读到的部分
可写区域[writerIndex, capacity)还能写入多少字节，写满之后会自动扩容

需要注意的是极限的情况：readerIndex 刚好等于 writerIndex，此时没有可读数据。如果 readerIndex 超过了 writerIndex，Netty 会抛出 IndexOutOfBoundsException。

示例代码：

package cn.juzicoding.netty.base;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.CharsetUtil;

/**
 * Netty ByteBuf 示例
 *
 * @author 橘子coding
 */
public class NettyByteBuf {
    public static void main(String[] args) {
        // 创建 byteBuf 对象，该对象内部包含一个字节数组 byte[10]
        // 通过 readerIndex、writerIndex、capacity 将 buffer 分成三个区域
        // 已经读取的区域：[0, readerIndex)
        // 可读取的区域：[readerIndex, writerIndex)
        // 可写的区域：[writerIndex, capacity)
        ByteBuf byteBuf = Unpooled.buffer(10);
        System.out.println("byteBuf=" + byteBuf);

        for (int i = 0; i < 8; i++) {
            byteBuf.writeByte(i);
        }
        System.out.println("byteBuf=" + byteBuf);

        for (int i = 0; i < 5; i++) {
            System.out.println(byteBuf.getByte(i));
        }
        System.out.println("byteBuf=" + byteBuf);

        for (int i = 0; i < 5; i++) {
            System.out.println(byteBuf.readByte());
        }
        System.out.println("byteBuf=" + byteBuf);


        // 用 Unpooled 工具类创建 ByteBuf
        ByteBuf byteBuf2 = Unpooled.copiedBuffer("hello,juzicoding!", CharsetUtil.UTF_8);
        // 使用相关的方法
        if (byteBuf2.hasArray()) {
            byte[] content = byteBuf2.array();
            // 将 content 转成字符串
            System.out.println(new String(content, CharsetUtil.UTF_8));
            System.out.println("byteBuf2=" + byteBuf2);

            // 获取数组 0 这个位置的字符 h 的 ASCII 码，h=104
            System.out.println(byteBuf2.getByte(0));

            // 可读的字节数：17
            int len = byteBuf2.readableBytes();
            System.out.println("len=" + len);

            // 使用 for 循环取出各个字节
            for (int i = 0; i < len; i++) {
                System.out.println((char) byteBuf2.getByte(i));
            }

            // 范围读取
            System.out.println(byteBuf2.getCharSequence(0, 6, CharsetUtil.UTF_8));
            System.out.println(byteBuf2.getCharSequence(6, 11, CharsetUtil.UTF_8));
        }
    }
}

运行结果（只保留索引变化相关的部分）：

byteBuf=UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf(ridx: 0, widx: 0, cap: 10)
byteBuf=UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf(ridx: 0, widx: 8, cap: 10)
0
1
2
3
4
byteBuf=UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf(ridx: 0, widx: 8, cap: 10)
0
1
2
3
4
byteBuf=UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf(ridx: 5, widx: 8, cap: 10)
hello,juzicoding!
byteBuf2=UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf(ridx: 0, widx: 17, cap: 17)
104
len=17
h
e
l
l
o
,
j
u
z
i
c
o
d
i
n
g
!
hello,
juzicoding!

两次连续的 0 到 4 值得对比一下：第一次是 getByte(i) 随机读，readerIndex 一直是 0；第二次是 readByte() 顺序读，读完 readerIndex 变成了 5。这就是 getXxx() 和 readXxx() 的区别——前者只按索引取值，不改变读索引，后者读完会推进读索引。

最后和 JDK 的 ByteBuffer 对比一下：ByteBuffer 只有一个 position，读完想写、写完想读，都得先 flip() 或 clear() 切换模式；ByteBuf 把读索引和写索引分开，读写互不干扰，也就不需要翻转这一步。此外 ByteBuf 的容量可以按需自动扩容，还支持池化复用，这些都是它比 ByteBuffer 更顺手的地方。

7. 实现一个控制台版本聊天室

实现一个能跑的控制台聊天室：服务端接收多个客户端连接，把任意一个人的发言广播给其他人；客户端在控制台输入内容发送，同时把服务端推来的消息打印出来。

7.1 需求分析

先把要做的行为定清楚，一共 4 条：

需求具体表现涉及角色
需求 1服务端控制台展示所有用户发送的消息服务端
需求 2群组内用户消息都可见，一个人发言，聊天室里其他人都能收到服务端 + 客户端
需求 3用户上线、下线时给出提示，并统计当前在线人数服务端
需求 4消息区分发送者，本人看到 [我]：内容，其他人看到 [发送者标识]：内容客户端

第 2 条是整个例子的核心：服务端不能只在 read 回调里回一句话，而要对所有在线连接做一次广播。所以关键问题变成了「服务端怎么知道当前有哪些连接」。

7.2 整体设计

用户标识：用 ip:port 代替用户名

需求 3 和需求 4 都需要知道「这条消息是谁发的」，最省事的做法是直接用客户端的地址：

  • 服务端侧取 channel.remoteAddress()，即客户端的 ip:port，例如 /127.0.0.1:64571；
  • 客户端侧取 channel.localAddress()，是同一个值的本机视角。

这个方案不用维护额外的用户名映射，也能唯一定位到一条连接。代价是可读性差，真实聊天室通常会让客户端先发一条昵称消息，服务端再用一个 Map<Channel, String> 保存昵称。

在线用户列表：ChannelGroup

服务端把在线连接放在一个静态常量里：

private static final ChannelGroup CHANNELS =
        new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

ChannelGroup 是 Netty 提供的一组 Channel 的集合，它有两个刚好对得上需求的能力：

  • 支持对组内所有 Channel 批量写，广播时不用自己维护并发集合；
  • 通道关闭时会自动把该 Channel 从组里移除，所以它可以直接当在线用户列表用，CHANNELS.size() 就是当前在线人数，不需要在 channelInactive 里手动 remove。

GlobalEventExecutor.INSTANCE 是 ChannelGroup 的构造参数，它提供一个全局的单线程执行器。按第 5 节说的「一个 Channel 的事件永远由同一个线程处理」，不同客户端可能落在不同的 worker 线程上，所以这个被多线程共享的集合需要一个能保证线程安全的执行器来兜底。

上下线时序

上线在 channelActive 里处理，三步的顺序不能调：

  1. 先给该客户端自己发一条欢迎语——此时它还没加入 CHANNELS，所以自己不会收到自己的上线广播；
  2. 再把上线提示广播给其他人；
  3. 最后 CHANNELS.add(channel)，从这一刻起它才能收到群聊消息。

下线在 channelInactive 里处理。这里有个容易踩的坑：触发这个回调时通道已经关闭，ChannelGroup 已经把它自动移除了，所以只需要广播一条下线提示，不能也不需要再手动 remove。

消息格式约定

服务端广播统一用 [发送者标识]：内容 这一种格式，客户端据此做两种判断：

  • 取第一个 ] 之前的内容，如果等于本机的 localAddress，说明这条是自己发的，显示成 [我]：内容；
  • 没有 [标识] 前缀的消息（比如欢迎语）原样打印。

用一套格式同时承载「谁发的」和「内容是什么」，客户端不用额外的状态就能区分自己和他人的消息。

内存管理：release 和新建 ByteBuf

第 6 节提过 ByteBuf 支持池化复用，代价是入站消息需要我们负责释放，这段代码里有两处对应处理：

  • channelRead 里的 msg 由 Netty 交给我们，用 try/finally 显式 buf.release()，否则会内存泄漏；
  • 广播和回显时每次新建 Unpooled.copiedBuffer(...)，它的生命周期交给 Netty——写完之后由 Netty 释放。同一个 ByteBuf 实例不能同时写给多个 Channel，这也正是广播时要一个一个新建的原因。

7.3 服务端实现

服务端分两个类：ChatServer 负责启动和参数配置，ChatServerHandler 负责聊天室业务。

package cn.juzicoding.netty.chat;

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.MultiThreadIoEventLoopGroup;
import io.netty.channel.nio.NioIoHandler;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * 聊天室服务端
 * <p>
 * 负责接收所有客户端的连接，再把任意一个客户端发送的消息广播给聊天室内的其他客户端，
 * 同时把收到的消息打印到服务端控制台。
 *
 * @author 橘子coding
 */
public class ChatServer {

    public static void main(String[] args) throws Exception {

        // 创建两个线程组 bossGroup 和 workerGroup
        // bossGroup 只处理连接请求，真正的和客户端业务处理，会交给 workerGroup 完成
        EventLoopGroup bossGroup = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory());
        EventLoopGroup workerGroup = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
        try {
            // 创建服务器端的启动对象
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    // 使用 NioServerSocketChannel 作为服务器的通道实现
                    .channel(NioServerSocketChannel.class)
                    // 服务端处理客户端连接请求是顺序处理的，同一时间只能处理一个客户端连接
                    // 多个客户端同时来的时候，服务端将不能处理的连接请求放在队列中等待处理
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    // 创建通道初始化对象，设置初始化参数
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // 聊天室的业务逻辑全部集中在 ChatServerHandler 中处理
                            ch.pipeline().addLast(new ChatServerHandler());
                        }
                    });
            System.out.println("chat server start，监听端口 " + ChatServerHandler.PORT + "。。");
            // 绑定端口并同步等待绑定完成
            ChannelFuture cf = bootstrap.bind(ChatServerHandler.PORT).sync();
            // 对通道关闭进行监听，这里会阻塞等待通道关闭完成
            cf.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

结构上和第 3 节的 NettyServer 完全一致，只有两处差别：端口抽成了 ChatServerHandler.PORT 常量，让客户端可以复用它；pipeline 里挂的是聊天室的 Handler。

package cn.juzicoding.netty.chat;

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.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * 聊天室服务端 Handler
 * <p>
 * 维护所有在线客户端的连接，负责用户上线/下线的提示以及群聊消息的广播。
 *
 * @author 橘子coding
 */
public class ChatServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 服务端监听端口
     */
    public static final int PORT = 9000;

    /**
     * 所有在线客户端的通道集合，使用 GlobalEventExecutor 保证自定义线程也能安全地操作该集合
     * <p>
     * ChannelGroup 会在通道关闭时自动把通道从集合中移除，因此可以作为在线用户列表使用
     */
    private static final ChannelGroup CHANNELS = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * 客户端与服务端建立连接后触发，此时用户上线
     *
     * @param ctx 上下文对象，含有通道 channel，管道 pipeline
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        // 用客户端的地址（ip:端口）作为用户标识，方便区分消息来自哪个客户端
        String user = channel.remoteAddress().toString();
        // 先向该客户端自己发送欢迎信息，此时通道还未加入 CHANNELS
        channel.writeAndFlush(Unpooled.copiedBuffer("欢迎 [" + user + "] 加入聊天室，直接输入内容回车即可发言", CharsetUtil.UTF_8));
        // 再把上线提示广播给聊天室内的其他用户
        broadcastExcept(channel, "[" + user + "] 上线了");
        // 加入在线用户列表，之后就能收到群聊消息
        CHANNELS.add(channel);
        System.out.println("用户上线：" + user + "（当前在线人数：" + CHANNELS.size() + "）");
    }

    /**
     * 读取客户端发送的数据
     *
     * @param ctx 上下文对象，含有通道 channel，管道 pipeline
     * @param msg 就是客户端发送的数据
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Channel channel = ctx.channel();
        ByteBuf buf = (ByteBuf) msg;
        try {
            String content = buf.toString(CharsetUtil.UTF_8).trim();
            // 客户端输入空行时不做任何处理
            if (content.isEmpty()) {
                return;
            }
            String user = channel.remoteAddress().toString();
            // 需求 1：服务端展示所有用户发送的消息
            System.out.println("[" + user + "]：" + content);
            // 需求 2：群组内用户消息都可见；需求 4：发送者本人也会收到自己那条消息的回显，
            // 由客户端把它显示成 [我]：xxx，其他客户端显示成 [ip:端口]：xxx
            String message = "[" + user + "]：" + content;
            broadcastExcept(channel, message);
            channel.writeAndFlush(Unpooled.copiedBuffer(message, CharsetUtil.UTF_8));
        } finally {
            // 入站消息需要我们负责释放，避免 ByteBuf 内存泄漏
            buf.release();
        }
    }

    /**
     * 客户端与服务端断开连接后触发，此时用户下线
     *
     * @param ctx 上下文对象，含有通道 channel，管道 pipeline
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        String user = channel.remoteAddress().toString();
        // ChannelGroup 已经自动把关闭的通道从集合中移除了，这里只需要把下线提示广播给其他用户
        // 自己不会再收到这条提示，客户端在连接断开时会有本地的“已断开”提示
        broadcastExcept(channel, "[" + user + "] 下线了");
        System.out.println("用户下线：" + user + "（当前在线人数：" + CHANNELS.size() + "）");
    }

    /**
     * 处理异常，一般是需要关闭通道
     *
     * @param ctx   上下文对象
     * @param cause 异常信息
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("客户端 " + ctx.channel().remoteAddress() + " 异常：" + cause.getMessage());
        ctx.close();
    }

    /**
     * 向除指定通道以外的所有在线客户端广播消息
     *
     * @param except  需要排除的通道，一般为消息发送者本人
     * @param message 消息内容
     */
    private void broadcastExcept(Channel except, String message) {
        for (Channel channel : CHANNELS) {
            if (channel != except && channel.isActive()) {
                // 每次发送都新建一个 ByteBuf，交给 Netty 在写完后负责释放
                channel.writeAndFlush(Unpooled.copiedBuffer(message, CharsetUtil.UTF_8));
            }
        }
    }
}

几个实现细节值得单独说一下：

  • channelRead 里对空行直接 return，但因为 return 在 try 块内，finally 里的 release() 依然会执行，不会漏释放；
  • 客户端发来的原始文本，服务端会先按 UTF-8 拼上 [发送者]： 再广播，所以同一条消息在服务端、发送者和接收者三处看到的字符串是一致的，只是客户端渲染时会把自己那条换成 [我]；
  • broadcastExcept 里的 channel != except 是需求设计的需要（发送者的回显单独发，避免收到两条），channel.isActive() 则是兜底：刚被动关闭、还来不及被 ChannelGroup 清理的通道不再写；
  • 广播和回显都用 writeAndFlush，前者写给其他连接，后者写给发送者本人，合起来就覆盖了聊天室里的所有人。

7.4 客户端实现

客户端同样是启动类加 Handler 的组合，区别在于它多了一条「读控制台输入」的职责。

package cn.juzicoding.netty.chat;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.MultiThreadIoEventLoopGroup;
import io.netty.channel.nio.NioIoHandler;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.CharsetUtil;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;

/**
 * 聊天室客户端
 * <p>
 * 启动后连接聊天室服务端，并在控制台读取用户输入发送到服务端；
 * 服务端推送过来的消息由 {@link ChatClientHandler} 打印到控制台。
 *
 * @author 橘子coding
 */
public class ChatClient {

    public static void main(String[] args) throws Exception {
        // 可以指定服务端地址，默认连接本机
        String host = args.length > 0 ? args[0] : "127.0.0.1";
        // 客户端需要一个事件循环组
        EventLoopGroup group = new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
        try {
            // 创建客户端启动对象，注意客户端使用的不是 ServerBootstrap 而是 Bootstrap
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    // 使用 NioSocketChannel 作为客户端的通道实现
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            // 加入处理器，负责显示服务端广播过来的消息
                            channel.pipeline().addLast(new ChatClientHandler());
                        }
                    });
            System.out.println("chat client start，正在连接 " + host + ":" + ChatServerHandler.PORT);
            // 连接服务端，sync 等待连接完成
            ChannelFuture channelFuture = bootstrap.connect(host, ChatServerHandler.PORT).sync();
            Channel channel = channelFuture.channel();
            // 主线程负责把控制台输入的内容发送给服务端，连接断开时结束循环
            readConsoleAndSend(channel);
            // 控制台输入结束（如 Ctrl+D）后主动关闭连接，结束本次聊天
            channel.close();
            // 对关闭通道进行监听
            channel.closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * 读取控制台输入并发送给服务端
     *
     * @param channel 与服务端建立的通道
     */
    private static void readConsoleAndSend(Channel channel) throws Exception {
        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in, StandardCharsets.UTF_8));
        String content;
        while (channel.isActive() && (content = reader.readLine()) != null) {
            // 直接发送原始文本（Netty 只接受 ByteBuf/FileRegion，因此这里需要包装成 ByteBuf），
            // 由服务端负责加上发送者标识后再广播
            channel.writeAndFlush(Unpooled.copiedBuffer(content, CharsetUtil.UTF_8));
        }
        if (!channel.isActive()) {
            System.out.println("与服务端的连接已断开，程序退出");
        }
    }
}
  • 客户端用的是 Bootstrap + NioSocketChannel，且支持用 args[0] 指定服务端地址，默认连本机；
  • 主线程被 readLine() 阻塞在控制台输入上，收到一行就 writeAndFlush 发出去，接收服务端广播的工作交给 IO 线程，两条线程各管一件事；
  • 发送前必须把字符串包装成 ByteBuf，因为 Netty 的写操作只接受 ByteBuf 或 FileRegion；
  • 循环条件里的 channel.isActive() 保证连接断开后能跳出循环，不会一直卡在控制台输入上。
package cn.juzicoding.netty.chat;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

/**
 * 聊天室客户端 Handler
 * <p>
 * 负责把服务端推送的消息打印到控制台，并按需求区分自己发送的消息和其他用户发送的消息。
 *
 * @author 橘子coding
 */
public class ChatClientHandler extends ChannelInboundHandlerAdapter {

    /**
     * 群聊消息的固定格式：[标识]：消息内容
     */
    private static final String MESSAGE_PREFIX = "[";

    /**
     * 本客户端的地址标识（ip:端口），由 Netty 的 IO 线程写入、主线程读取，因此使用 volatile 修饰
     */
    private volatile String localAddress;

    /**
     * 当客户端连接服务器完成就会触发该方法
     *
     * @param ctx 上下文对象，含有通道 channel，管道 pipeline
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 连接建立后本地地址就已经确定，可用于判断消息是不是自己发送的
        localAddress = ctx.channel().localAddress().toString();
    }

    /**
     * 当通道有读取事件时会触发，即服务端发送数据给客户端
     *
     * @param ctx 上下文对象，含有通道 channel，管道 pipeline
     * @param msg 服务端发送的数据
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 将 msg 转成一个 ByteBuf，类似 NIO 的 ByteBuffer
        ByteBuf buf = (ByteBuf) msg;
        try {
            String message = buf.toString(CharsetUtil.UTF_8);
            int end = message.indexOf(']');
            // 没有 [标识] 前缀的消息（如欢迎信息）以及格式异常的消息直接原样打印
            if (!message.startsWith(MESSAGE_PREFIX) || end < 0) {
                System.out.println(message);
                return;
            }
            // 需求 4：自己发送的消息显示为 [我]：xxx，其他用户发送的消息显示为 [ip:端口]：xxx
            // 服务端广播格式为 "[发送者地址]：消息内容"，地址与本地地址一致说明这条消息是自己发的
            String sender = message.substring(1, end);
            if (sender.equals(localAddress)) {
                System.out.println("[我]" + message.substring(end + 1));
            } else {
                System.out.println(message);
            }
        } finally {
            // 入站消息需要我们负责释放，避免 ByteBuf 内存泄漏
            buf.release();
        }
    }

    /**
     * 与服务端断开连接后触发，此时结束进程，避免主线程一直阻塞在控制台输入上
     *
     * @param ctx 上下文对象，含有通道 channel，管道 pipeline
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("已断开与聊天室服务端的连接");
        // 非守护线程全部结束后 JVM 才会退出，这里直接结束进程以便关闭控制台输入线程
        System.exit(0);
    }

    /**
     * 处理异常，一般是需要关闭通道
     *
     * @param ctx   上下文对象
     * @param cause 异常信息
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("客户端异常：" + cause.getMessage());
        ctx.close();
    }
}
  • MESSAGE_PREFIX 配合 indexOf(']') 做格式判断：既认出了群聊消息，也放过了欢迎语这类没有前缀的系统消息；
  • localAddress 由 IO 线程在 channelActive 里写入、主线程可能读取，所以用 volatile 修饰，保证可见性；
  • channelInactive 里的 System.exit(0) 是为了解决「主线程还阻塞在 readLine() 上、进程不退出」的问题。正常路径下主动关闭连接时，readConsoleAndSend 的循环会因为 isActive() 为 false 而跳出，同样会打印「已断开与聊天室服务端的连接」。

7.5 运行演示

先启动 ChatServer，再启动任意多个 ChatClient（需要开多个终端，或在 IDE 里勾选 Allow multiple instances）。下面是一次真实运行的输出：A 先上线发言，B 随后加入并回复，最后 A 再发一条。

服务端控制台：

chat server start，监听端口 9000。。
用户上线：/127.0.0.1:64571（当前在线人数：1）
[/127.0.0.1:64571]：大家好，我是 A
用户上线：/127.0.0.1:64573（当前在线人数：2）
[/127.0.0.1:64573]：你好 A，我是 B
[/127.0.0.1:64571]：收到，先撤了
用户下线：/127.0.0.1:64573（当前在线人数：1）
用户下线：/127.0.0.1:64571（当前在线人数：0）

客户端 A：

chat client start，正在连接 127.0.0.1:9000
欢迎 [/127.0.0.1:64571] 加入聊天室，直接输入内容回车即可发言
[我]：大家好，我是 A
[/127.0.0.1:64573] 上线了
[/127.0.0.1:64573]：你好 A，我是 B
[我]：收到，先撤了
[/127.0.0.1:64573] 下线了
已断开与聊天室服务端的连接

客户端 B：

chat client start，正在连接 127.0.0.1:9000
欢迎 [/127.0.0.1:64573] 加入聊天室，直接输入内容回车即可发言
[我]：你好 A，我是 B
[/127.0.0.1:64571]：收到，先撤了
已断开与聊天室服务端的连接

对照服务端和两个客户端的输出，几个设计点都能得到验证：

  • 服务端能看到所有人的消息，且在 A 只有一个人在线时也在正常记录；
  • B 上线后 A 收到「上线了」，A 能看到 B 的消息，B 也能看到 A 后续的消息，说明广播生效；
  • 同一条消息在 A 那里显示成 [我]：，在 B 那里显示成 [/127.0.0.1:64571]：，需求 4 成立；
  • 服务端的在线人数从 1 涨到 2，B 断开后又回到 1，最后归零，说明 ChannelGroup 的自动移除确实生效了；
  • B 一直没收到「自己下线」的提示，和 channelInactive 里排除自己的设计一致。
编辑此页
最近更新:
贡献者: juzicoding