核心概念说明

Netty 是什么？

Netty 是一个基于 NIO（非阻塞 I/O）的高性能客户端-服务器端编程框架，它极大地简化了 TCP、UDP 套接字服务器等网络应用程序的开发过程。

Reactor 线程模型

Netty 采用 Reactor 线程模型来处理 I/O 操作，主要包含三种模式：

  1. 单线程 Reactor 模型 - 所有 I/O 操作在单个线程中处理
  2. 多线程 Reactor 模型 - 使用线程池处理 I/O 操作
  3. 主从 Reactor 多线程模型 - 本示例采用的模型，主 Reactor 负责接收连接，从 Reactor 负责处理 I/O 读写

开发环境配置

  • Java 17
  • SpringBoot 3.5.5（实际本示例代码在我的一个子模块中，所以示例中未直接使用 SpringBoot，不影响运行）
  • IDEA 2024.1
  • Netty 4.1.108.Final

代码模块结构

核心组件说明

  • NettyServer：服务器端实现，监听指定端口并处理客户端连接
  • NettyClient：客户端实现，连接到服务器并发送消息
  • ServerHandler：服务器端处理器，处理客户端连接和消息
  • ClientHandler：客户端处理器，处理服务器响应
  • NettyConfig：Netty 相关配置常量

完整代码实现

一、Maven 依赖配置 (pom.xml)

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>netty-example</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <!-- Netty 核心依赖 -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.108.Final</version>
        </dependency>

        <!-- 日志依赖 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.36</version>
        </dependency>

        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.4.12</version>
        </dependency>
        
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-core</artifactId>
            <version>1.4.12</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.11.0</version>
                <configuration>
                    <source>17</source>
                    <target>17</target>
                </configuration>
            </plugin>
            
            <!-- 添加 Maven Shade 插件，用于打包可执行的 JAR 文件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.2.4</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.example.netty.NettyServer</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

二、Java 核心代码实现

1. 配置类 - NettyConfig.java
/**
 * Netty 配置常量类
 * 集中管理服务器和客户端的配置参数
 */
public class NettyConfig {
    /**
     * 服务器默认端口号
     */
    public static final int DEFAULT_PORT = 8080;
    
    /**
     * 服务器默认主机地址
     */
    public static final String DEFAULT_HOST = "localhost";
    
    /**
     * 服务器监听队列大小（等待处理的连接请求数）
     */
    public static final int SO_BACKLOG = 128;
}
2. 服务器端 - NettyServer.java
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;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Netty 服务器端实现
 * 采用主从 Reactor 多线程模型处理客户端连接和网络 I/O
 */
public class NettyServer {
    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
    
    // 用于处理接收客户端连接的线程组（主Reactor）
    private EventLoopGroup bossGroup;
    
    // 用于处理网络读写操作的线程组（从Reactor）
    private EventLoopGroup workerGroup;
    
    // 服务器启动引导类
    private ServerBootstrap serverBootstrap;
    
    // 服务器通道未来对象，用于异步操作
    private ChannelFuture channelFuture;
    
    /**
     * 初始化服务器资源
     * 创建线程组和配置服务器启动参数
     */
    public NettyServer() {
        // 创建主Reactor线程组，用于接收客户端连接
        bossGroup = new NioEventLoopGroup(1);
        
        // 创建从Reactor线程组，用于处理客户端连接的读写操作
        workerGroup = new NioEventLoopGroup();
        
        // 创建服务器启动引导类实例
        serverBootstrap = new ServerBootstrap();
        
        // 配置服务器启动参数
        serverBootstrap.group(bossGroup, workerGroup)  // 设置主从线程组
                .channel(NioServerSocketChannel.class)  // 设置通道类型为NIO服务器通道
                .option(ChannelOption.SO_BACKLOG, NettyConfig.SO_BACKLOG)  // 设置TCP连接队列大小
                .childOption(ChannelOption.SO_KEEPALIVE, true)  // 设置保持连接状态
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        // 添加编解码器和业务处理器
                        ch.pipeline().addLast(new StringDecoder());  // 字符串解码器
                        ch.pipeline().addLast(new StringEncoder());  // 字符串编码器
                        ch.pipeline().addLast(new ServerHandler());  // 业务处理器
                    }
                });
    }
    
    /**
     * 启动服务器
     * @param port 服务器监听端口号
     * @throws InterruptedException 当线程被中断时抛出
     */
    public void start(int port) throws InterruptedException {
        try {
            // 绑定端口并启动服务器
            channelFuture = serverBootstrap.bind(port).sync();
            System.out.println("Netty服务器启动成功，监听端口: " + port);
            logger.info("Netty服务器启动成功，监听端口: {}", port);
            
            // 等待服务器通道关闭
            channelFuture.channel().closeFuture().sync();
        } finally {
            // 释放资源
            close();
        }
    }
    
    /**
     * 关闭服务器并释放资源
     */
    public void close() {
        if (channelFuture != null) {
            channelFuture.channel().close();
        }
        
        // 优雅地关闭主Reactor线程组
        if (bossGroup != null) {
            bossGroup.shutdownGracefully();
        }
        
        // 优雅地关闭从Reactor线程组
        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
        }
        
        System.out.println("Netty服务器已关闭");
        logger.info("Netty服务器已关闭");
    }
    
    /**
     * 主方法，用于启动服务器示例
     * @param args 命令行参数
     */
    public static void main(String[] args) {
        NettyServer server = new NettyServer();
        try {
            server.start(NettyConfig.DEFAULT_PORT);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("服务器启动被中断");
            logger.error("服务器启动被中断", e);
        } catch (Exception e) {
            System.err.println("服务器启动失败: " + e.getMessage());
            logger.error("服务器启动失败", e);
        }
    }
}
3. 服务器处理器 - ServerHandler.java
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * 服务器端处理器
 * 处理客户端连接、消息接收和异常情况
 */
public class ServerHandler extends ChannelInboundHandlerAdapter {
    private static final Logger logger = LoggerFactory.getLogger(ServerHandler.class);
    
    /**
     * 当客户端连接建立时调用
     * @param ctx 通道处理上下文
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("客户端连接建立: " + ctx.channel().remoteAddress());
        logger.info("客户端连接建立: {}", ctx.channel().remoteAddress());
    }
    
    /**
     * 当接收到客户端消息时调用
     * @param ctx 通道处理上下文
     * @param msg 接收到的消息
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 接收客户端发送的消息
        String message = (String) msg;
        String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss:SSS"));
        System.out.println("接收到客户端消息: " + message + "，收到消息时间：" + time);
        logger.info("接收到客户端消息: 【{}】", message);
        
        // 向客户端发送响应消息
        String response = "我是服务器，已收到消息: " + message;
        ctx.writeAndFlush(response);
    }
    
    /**
     * 通道读取完成时调用
     * @param ctx 通道处理上下文
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        // 刷新缓冲区，确保所有消息都发送出去
        ctx.flush();
    }
    
    /**
     * 处理通道异常
     * @param ctx 通道处理上下文
     * @param cause 异常信息
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.err.println("服务器处理异常: " + cause.getMessage());
        logger.error("服务器处理异常", cause);
        // 关闭通道
        ctx.close();
    }
}
4. 客户端 - NettyClient.java
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;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * Netty 客户端示例
 * 实现了一个简单的客户端，能够连接到服务器并发送消息
 */
public class NettyClient {
    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);

    // 用于处理网络操作的线程组
    private EventLoopGroup workerGroup;

    // 客户端启动引导类
    private Bootstrap bootstrap;

    // 客户端通道未来对象
    private ChannelFuture channelFuture;

    /**
     * 初始化客户端资源
     * 创建线程组和配置客户端启动参数
     */
    public NettyClient() {
        // 创建工作线程组
        workerGroup = new NioEventLoopGroup();

        // 创建客户端启动引导类实例
        bootstrap = new Bootstrap();

        // 配置客户端启动参数
        bootstrap.group(workerGroup)  // 设置线程组
                .channel(NioSocketChannel.class)  // 设置通道类型为NIO客户端通道
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        // 添加编解码器和业务处理器
                        ch.pipeline().addLast(new StringDecoder());  // 字符串解码器
                        ch.pipeline().addLast(new StringEncoder());  // 字符串编码器
                        ch.pipeline().addLast(new ClientHandler());  // 业务处理器
                    }
                });
    }

    /**
     * 连接到服务器
     *
     * @param host 服务器主机地址
     * @param port 服务器端口号
     * @throws InterruptedException 当线程被中断时抛出
     */
    public void connect(String host, int port) throws InterruptedException {
        try {
            // 连接到服务器
            channelFuture = bootstrap.connect(host, port).sync();
            System.out.println("Netty客户端连接成功，服务器地址: " + host + ":" + port);
            logger.info("Netty客户端连接成功，服务器地址: {}:{}", host, port);

            // 等待通道关闭
            channelFuture.channel().closeFuture().sync();
        } finally {
            // 释放资源
            close();
        }
    }

    /**
     * 发送消息到服务器
     *
     * @param msg 要发送的消息
     */
    public void sendMsg(String msg) {
        if (channelFuture != null && channelFuture.channel().isActive()) {
            channelFuture.channel().writeAndFlush(msg);
        } else {
            System.out.println("通道未激活，无法发送消息");
            logger.warn("通道未激活，无法发送消息");
        }
    }

    /**
     * 关闭客户端并释放资源
     */
    public void close() {
        if (channelFuture != null) {
            channelFuture.channel().close();
        }

        // 优雅地关闭线程组
        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
        }

        System.out.println("Netty客户端已关闭");
        logger.info("Netty客户端已关闭");
    }

    /**
     * 主方法，用于启动客户端示例
     *
     * @param args 命令行参数
     */
    public static void main(String[] args) {
        NettyClient client = new NettyClient();
        try {
            // 在新线程中启动客户端连接
            new Thread(() -> {
                try {
                    client.connect(NettyConfig.DEFAULT_HOST, NettyConfig.DEFAULT_PORT);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    System.err.println("客户端连接被中断");
                    logger.error("客户端连接被中断", e);
                } catch (Exception e) {
                    System.err.println("客户端连接失败: " + e.getMessage());
                    logger.error("客户端连接失败", e);
                }
            }).start();

            // 等待连接建立后发送消息
            Thread.sleep(2000);
            client.sendMsg("Hello, Netty!");
            for (int i = 0; i < 50; i++) {
                Thread.sleep(2000);
                String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss:SSS"));
                client.sendMsg("Message " + i + "-时间：" + time);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("客户端操作被中断");
            logger.error("客户端操作被中断", e);
        }
    }
}
5. 客户端处理器 - ClientHandler.java
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * 客户端处理器
 * 处理服务器响应消息和异常情况
 */
public class ClientHandler extends ChannelInboundHandlerAdapter {
    private static final Logger logger = LoggerFactory.getLogger(ClientHandler.class);
    
    /**
     * 当与服务器连接建立时调用
     * @param ctx 通道处理上下文
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss:SSS"));
        System.out.println("与服务器连接建立: " + ctx.channel().remoteAddress() + "，连接时间：" + time);
        logger.info("与服务器连接建立: {}", ctx.channel().remoteAddress());
    }
    
    /**
     * 当接收到服务器消息时调用
     * @param ctx 通道处理上下文
     * @param msg 接收到的消息
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 接收服务器发送的响应消息
        String message = (String) msg;
        String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss:SSS"));
        System.out.println("接收到服务器响应: " + message + "，收到响应时间：" + time);
        logger.info("接收到服务器响应: {}", message);
    }
    
    /**
     * 处理通道异常
     * @param ctx 通道处理上下文
     * @param cause 异常信息
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.err.println("客户端处理异常: " + cause.getMessage());
        logger.error("客户端处理异常", cause);
        // 关闭通道
        ctx.close();
    }
}

三、日志配置文件

在 resources 目录下创建 logback.xml：

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <!-- 控制台输出配置 -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>

    <!-- 文件输出配置 -->
    <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>logs/netty-example.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>logs/netty-example.%d{yyyy-MM-dd}.log</fileNamePattern>
            <maxHistory>30</maxHistory>
        </rollingPolicy>
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>

    <!-- 根日志级别配置 -->
    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
        <appender-ref ref="FILE"/>
    </root>
</configuration>

运行步骤

执行顺序

  1. 启动服务器：首先运行 NettyServer 的 main 方法启动服务器
  2. 启动客户端：然后运行 NettyClient 的 main 方法启动客户端并自动发送消息

预期输出结果

服务器端控制台输出：
Netty服务器启动成功，监听端口: 8080
客户端连接建立: /127.0.0.1:65115
接收到客户端消息: Hello, Netty!，收到消息时间：09:30:01:932
接收到客户端消息: Message 0-时间：09:30:03:879，收到消息时间：09:30:03:879
接收到客户端消息: Message 1-时间：09:30:05:893，收到消息时间：09:30:05:893
...
接收到客户端消息: Message 47-时间：09:31:38:282，收到消息时间：09:31:38:284
接收到客户端消息: Message 48-时间：09:31:40:283，收到消息时间：09:31:40:285
接收到客户端消息: Message 49-时间：09:31:42:289，收到消息时间：09:31:42:291
客户端控制台输出：
与服务器连接建立: localhost/127.0.0.1:8080，连接时间：09:30:00:174
Netty客户端连接成功，服务器地址: localhost:8080
接收到服务器响应: 我是服务器，已收到消息: Hello, Netty!，收到响应时间：09:30:01:948
接收到服务器响应: 我是服务器，已收到消息: Message 0-时间：09:30:03:879，收到响应时间：09:30:03:879
接收到服务器响应: 我是服务器，已收到消息: Message 1-时间：09:30:05:893，收到响应时间：09:30:05:893
...
接收到服务器响应: 我是服务器，已收到消息: Message 47-时间：09:31:38:282，收到响应时间：09:31:38:286
接收到服务器响应: 我是服务器，已收到消息: Message 48-时间：09:31:40:283，收到响应时间：09:31:40:287
接收到服务器响应: 我是服务器，已收到消息: Message 49-时间：09:31:42:289，收到响应时间：09:31:42:292

技术要点总结

核心组件说明

  1. EventLoopGroup：事件循环组，负责处理所有的 I/O 事件
  2. ServerBootstrap/Bootstrap：服务器/客户端启动引导类
  3. ChannelPipeline：责任链模式的处理管道
  4. ChannelHandler：业务处理器，处理具体的 I/O 事件

主从 Reactor 模型优势

  • 主 Reactor：专门处理连接请求，提高连接处理效率
  • 从 Reactor：专门处理 I/O 读写，提高数据处理效率
  • 线程资源优化：避免单线程瓶颈，充分利用多核 CPU

异常处理机制

  • 使用 exceptionCaught 方法统一处理异常
  • 优雅的资源释放和连接关闭
  • 完善的日志记录便于问题排查
# 网络 # java # 开发语言
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

为什么越智能的 AI Agent，越需要一层“笨系统“？

avatar AI Agent技术社区
cover

AI Agent工具调用踩坑：并行还是串行？一张表分清执行策略

avatar AI Agent技术社区

2026年企业AI Agent工具横评：Claude Code国内替代选型指南

最近为了给团队找适配国内办公环境的Claude Code替代方案，调研了AI办公赛道几款主流Agent产品，最终选择了飞书 aily，它的核心优势是完全基于国内企业日常协作场景设计，AI生成的所有内容都可以直接沉淀到现有工作流里，不需要额外做跨系统的适配和数据迁移。

avatar AI Agent技术社区