Netty 完全指南：构建高性能网络应用的终极秘籍

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

// EventLoop 是 Netty 的核心抽象
public class EventLoopExample {
    public static void main(String[] args) {
        // Boss EventLoopGroup：负责接受连接
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);

        // Worker EventLoopGroup：负责处理 I/O 事件
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) {
                     // 配置 pipeline
                 }
             });

            System.out.println("Boss EventLoopGroup 线程数: " +
                ((NioEventLoopGroup) bossGroup).executorCount());
            System.out.println("Worker EventLoopGroup 线程数: " +
                ((NioEventLoopGroup) workerGroup).executorCount());

        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

// 自定义 EventLoop 配置
public class EventLoopConfiguration {

    public static EventLoopGroup createOptimizedEventLoopGroup() {
        // 根据 CPU 核数确定线程数
        int threads = Runtime.getRuntime().availableProcessors() * 2;

        return new NioEventLoopGroup(threads, new ThreadFactory() {
            private final AtomicInteger counter = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("Netty-Worker-" + counter.incrementAndGet());
                thread.setDaemon(true);
                // 设置线程优先级
                thread.setPriority(Thread.MAX_PRIORITY);
                return thread;
            }
        });
    }

    // EventLoop 任务调度示例
    public static void scheduleTask(EventLoop eventLoop) {
        // 延迟执行任务
        eventLoop.schedule(() -> {
            System.out.println("延迟任务执行: " + Thread.currentThread().getName());
        }, 5, TimeUnit.SECONDS);

        // 周期性任务
        eventLoop.scheduleAtFixedRate(() -> {
            System.out.println("周期性任务: " + new Date());
        }, 0, 10, TimeUnit.SECONDS);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

// 自定义 ChannelHandler 演示完整生命周期
public class LifecycleChannelHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        System.out.println("Handler 被添加到 pipeline");
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) {
        System.out.println("Channel 注册到 EventLoop");
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("Channel 变为活跃状态，可以读写数据");

        // 发送欢迎消息
        String welcomeMsg = "欢迎连接到 Netty 服务器！\n";
        ctx.writeAndFlush(Unpooled.copiedBuffer(welcomeMsg, CharsetUtil.UTF_8));
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf buf = (ByteBuf) msg;
        try {
            String receivedData = buf.toString(CharsetUtil.UTF_8);
            System.out.println("接收到数据: " + receivedData);

            // 回显数据
            String response = "Echo: " + receivedData;
            ctx.writeAndFlush(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8));

        } finally {
            buf.release(); // 重要：释放 ByteBuf
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        System.out.println("Channel 读取完成");
        ctx.flush(); // 确保数据被发送
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        System.out.println("Channel 变为非活跃状态");
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) {
        System.out.println("Channel 从 EventLoop 注销");
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        System.out.println("Handler 从 pipeline 移除");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.err.println("异常处理: " + cause.getMessage());
        cause.printStackTrace();
        ctx.close(); // 关闭连接
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

// Pipeline 配置和管理
public class PipelineManager {

    public static void configurePipeline(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();

        // 1. 日志处理器（最前面）
        pipeline.addFirst("logger", new LoggingHandler(LogLevel.INFO));

        // 2. 空闲检测
        pipeline.addLast("idleState", new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS));

        // 3. 协议解码器
        pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(1024 * 1024, 0, 4, 0, 4));
        pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));

        // 4. 字符串编解码
        pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));

        // 5. 业务处理器
        pipeline.addLast("businessHandler", new BusinessHandler());

        // 6. 异常处理器（最后面）
        pipeline.addLast("exceptionHandler", new ExceptionHandler());
    }

    // 动态添加处理器
    public static void addSSLHandler(ChannelPipeline pipeline, SslContext sslContext) {
        pipeline.addFirst("ssl", sslContext.newHandler(pipeline.channel().alloc()));
    }

    // 动态移除处理器
    public static void removeHandler(ChannelPipeline pipeline, String name) {
        if (pipeline.get(name) != null) {
            pipeline.remove(name);
        }
    }
}

// 空闲状态处理
class IdleHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                System.out.println("读空闲，关闭连接");
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

// ByteBuf 详细使用示例
public class ByteBufExample {

    public static void basicOperations() {
        // 创建 ByteBuf
        ByteBuf buffer = Unpooled.buffer(256);

        System.out.println("初始状态:");
        printBufferInfo(buffer);

        // 写入数据
        buffer.writeBytes("Hello".getBytes());
        buffer.writeInt(42);
        buffer.writeLong(System.currentTimeMillis());

        System.out.println("写入数据后:");
        printBufferInfo(buffer);

        // 读取数据
        byte[] helloBytes = new byte[5];
        buffer.readBytes(helloBytes);
        int number = buffer.readInt();
        long timestamp = buffer.readLong();

        System.out.println("读取的数据:");
        System.out.println("String: " + new String(helloBytes));
        System.out.println("Int: " + number);
        System.out.println("Long: " + timestamp);

        System.out.println("读取数据后:");
        printBufferInfo(buffer);

        // 重要：释放 ByteBuf
        buffer.release();
    }

    private static void printBufferInfo(ByteBuf buf) {
        System.out.println("readerIndex: " + buf.readerIndex());
        System.out.println("writerIndex: " + buf.writerIndex());
        System.out.println("capacity: " + buf.capacity());
        System.out.println("readableBytes: " + buf.readableBytes());
        System.out.println("writableBytes: " + buf.writableBytes());
        System.out.println("---");
    }

    // 内存池使用
    public static void pooledByteBuf() {
        ByteBufAllocator allocator = PooledByteBufAllocator.DEFAULT;

        // 申请堆内存
        ByteBuf heapBuf = allocator.heapBuffer(256);

        // 申请直接内存（堆外内存）
        ByteBuf directBuf = allocator.directBuffer(256);

        try {
            // 使用缓冲区
            heapBuf.writeBytes("Heap Buffer".getBytes());
            directBuf.writeBytes("Direct Buffer".getBytes());

            System.out.println("Heap Buffer: " + heapBuf.toString(CharsetUtil.UTF_8));
            System.out.println("Direct Buffer: " + directBuf.toString(CharsetUtil.UTF_8));

        } finally {
            // 释放内存
            heapBuf.release();
            directBuf.release();
        }
    }

    // 复合缓冲区
    public static void compositeByteBuf() {
        CompositeByteBuf composite = Unpooled.compositeBuffer();

        ByteBuf header = Unpooled.copiedBuffer("Header", CharsetUtil.UTF_8);
        ByteBuf body = Unpooled.copiedBuffer("Body Content", CharsetUtil.UTF_8);

        // 将多个 ByteBuf 组合
        composite.addComponents(true, header, body);

        System.out.println("Composite Buffer: " + composite.toString(CharsetUtil.UTF_8));

        composite.release();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

// 内存泄漏检测配置
public class MemoryLeakDetection {

    static {
        // 设置内存泄漏检测级别
        System.setProperty("io.netty.leakDetection.level", "ADVANCED");
        // 可选值：DISABLED, SIMPLE, ADVANCED, PARANOID
    }

    // 正确的 ByteBuf 使用模式
    public static void correctByteBufUsage(ChannelHandlerContext ctx, Object msg) {
        ByteBuf buf = (ByteBuf) msg;
        try {
            // 处理数据
            String data = buf.toString(CharsetUtil.UTF_8);
            System.out.println("Received: " + data);

            // 创建响应
            ByteBuf response = ctx.alloc().buffer();
            try {
                response.writeBytes("Response".getBytes());
                ctx.writeAndFlush(response);
                response = null; // 防止在 finally 中重复释放
            } finally {
                if (response != null) {
                    response.release();
                }
            }

        } finally {
            buf.release(); // 确保释放输入缓冲区
        }
    }

    // 使用 ReferenceCountUtil 安全释放
    public static void safeByteBufUsage(Object msg) {
        try {
            if (msg instanceof ByteBuf) {
                ByteBuf buf = (ByteBuf) msg;
                String data = buf.toString(CharsetUtil.UTF_8);
                System.out.println("Data: " + data);
            }
        } finally {
            ReferenceCountUtil.release(msg); // 安全释放
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

// 完整的 Echo 服务器实现
public class EchoServer {
    private final int port;

    public EchoServer(int port) {
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 1024)
             .option(ChannelOption.SO_REUSEADDR, true)
             .childOption(ChannelOption.SO_KEEPALIVE, true)
             .childOption(ChannelOption.TCP_NODELAY, true)
             .childOption(ChannelOption.SO_RCVBUF, 32 * 1024)
             .childOption(ChannelOption.SO_SNDBUF, 32 * 1024)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) {
                     ChannelPipeline p = ch.pipeline();

                     // 添加处理器
                     p.addLast(new IdleStateHandler(60, 30, 0, TimeUnit.SECONDS));
                     p.addLast(new EchoServerHandler());
                 }
             });

            // 绑定端口并启动服务器
            ChannelFuture f = b.bind(port).sync();
            System.out.println("Echo Server started on port " + port);

            // 等待服务器 socket 关闭
            f.channel().closeFuture().sync();

        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        int port = args.length > 0 ? Integer.parseInt(args[0]) : 8080;
        new EchoServer(port).start();
    }
}

// Echo 服务器处理器
class EchoServerHandler extends ChannelInboundHandlerAdapter {
    private static final AtomicLong connectionCount = new AtomicLong(0);
    private static final AtomicLong messageCount = new AtomicLong(0);

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        long connections = connectionCount.incrementAndGet();
        System.out.println("Client connected: " + ctx.channel().remoteAddress() +
                          ", Total connections: " + connections);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        long connections = connectionCount.decrementAndGet();
        System.out.println("Client disconnected: " + ctx.channel().remoteAddress() +
                          ", Total connections: " + connections);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf in = (ByteBuf) msg;
        try {
            String message = in.toString(CharsetUtil.UTF_8);
            long msgCount = messageCount.incrementAndGet();

            System.out.println("Received message #" + msgCount + ": " + message.trim());

            // 构建响应
            String response = "Echo #" + msgCount + ": " + message;
            ByteBuf responseBuffer = Unpooled.copiedBuffer(response, CharsetUtil.UTF_8);

            ctx.writeAndFlush(responseBuffer);

        } finally {
            in.release();
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            String eventType = null;
            switch (event.state()) {
                case READER_IDLE:
                    eventType = "读空闲";
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;
            }
            System.out.println(ctx.channel().remoteAddress() + " 超时事件: " + eventType);
            ctx.channel().close();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.err.println("Server exception: " + cause.getMessage());
        cause.printStackTrace();
        ctx.close();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

// Echo 客户端实现
public class EchoClient {
    private final String host;
    private final int port;

    public EchoClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .option(ChannelOption.TCP_NODELAY, true)
             .option(ChannelOption.SO_KEEPALIVE, true)
             .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(new EchoClientHandler());
                 }
             });

            // 连接服务器
            ChannelFuture f = b.connect(host, port).sync();
            System.out.println("Connected to " + host + ":" + port);

            // 等待连接关闭
            f.channel().closeFuture().sync();

        } finally {
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        String host = args.length > 0 ? args[0] : "localhost";
        int port = args.length > 1 ? Integer.parseInt(args[1]) : 8080;

        new EchoClient(host, port).start();
    }
}

// Echo 客户端处理器
class EchoClientHandler extends ChannelInboundHandlerAdapter {
    private final AtomicInteger messageCounter = new AtomicInteger(0);

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("Connected to server: " + ctx.channel().remoteAddress());

        // 开始发送消息
        sendMessage(ctx);
    }

    private void sendMessage(ChannelHandlerContext ctx) {
        // 定时发送消息
        ctx.executor().schedule(() -> {
            if (ctx.channel().isActive()) {
                int msgNum = messageCounter.incrementAndGet();
                String message = "Hello Netty! Message #" + msgNum +
                               " at " + new Date() + "\n";

                ByteBuf buffer = Unpooled.copiedBuffer(message, CharsetUtil.UTF_8);
                ctx.writeAndFlush(buffer);

                // 继续发送下一条消息
                sendMessage(ctx);
            }
        }, 2, TimeUnit.SECONDS);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf in = (ByteBuf) msg;
        try {
            String response = in.toString(CharsetUtil.UTF_8);
            System.out.println("Server response: " + response.trim());
        } finally {
            in.release();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.err.println("Client exception: " + cause.getMessage());
        cause.printStackTrace();
        ctx.close();
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

// HTTP 服务器实现
public class HttpServer {
    private final int port;

    public HttpServer(int port) {
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 1024)
             .childOption(ChannelOption.SO_KEEPALIVE, true)
             .childOption(ChannelOption.TCP_NODELAY, true)
             .childHandler(new HttpServerInitializer());

            ChannelFuture f = b.bind(port).sync();
            System.out.println("HTTP Server started on http://localhost:" + port);

            f.channel().closeFuture().sync();

        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        int port = args.length > 0 ? Integer.parseInt(args[0]) : 8080;
        new HttpServer(port).start();
    }
}

// HTTP 服务器初始化器
class HttpServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    public void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();

        // HTTP 编解码器
        pipeline.addLast(new HttpServerCodec());

        // HTTP 对象聚合器
        pipeline.addLast(new HttpObjectAggregator(65536));

        // HTTP 内容压缩
        pipeline.addLast(new HttpContentCompressor());

        // 自定义业务处理器
        pipeline.addLast(new HttpServerHandler());
    }
}

// HTTP 服务器处理器
class HttpServerHandler extends ChannelInboundHandlerAdapter {
    private static final Map<String, String> routes = new HashMap<>();
    private static final AtomicLong requestCount = new AtomicLong(0);

    static {
        // 配置路由
        routes.put("/", "Welcome to Netty HTTP Server!");
        routes.put("/hello", "Hello, World!");
        routes.put("/time", "Current time: " + new Date());
        routes.put("/stats", "Request count: ");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof FullHttpRequest) {
            FullHttpRequest request = (FullHttpRequest) msg;

            long count = requestCount.incrementAndGet();
            String uri = request.uri();
            HttpMethod method = request.method();

            System.out.println("Request #" + count + ": " + method + " " + uri);

            // 处理请求
            handleHttpRequest(ctx, request, count);
        }
    }

    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request, long count) {
        String uri = request.uri();
        String responseContent;
        HttpResponseStatus status = HttpResponseStatus.OK;

        // 路由处理
        if (routes.containsKey(uri)) {
            responseContent = routes.get(uri);
            if (uri.equals("/stats")) {
                responseContent += count;
            }
        } else if (uri.equals("/api/echo")) {
            // API 接口：回显请求体
            ByteBuf content = request.content();
            responseContent = "Echo: " + content.toString(CharsetUtil.UTF_8);
        } else if (uri.startsWith("/api/user/")) {
            // RESTful API 示例
            String userId = uri.substring("/api/user/".length());
            responseContent = "{\"userId\": \"" + userId + "\", \"name\": \"User " + userId + "\"}";
        } else {
            // 404 处理
            responseContent = "404 Not Found: " + uri;
            status = HttpResponseStatus.NOT_FOUND;
        }

        // 构建 HTTP 响应
        sendHttpResponse(ctx, request, responseContent, status);
    }

    private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest request,
                                 String content, HttpResponseStatus status) {

        ByteBuf responseContent = Unpooled.copiedBuffer(content, CharsetUtil.UTF_8);

        FullHttpResponse response = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1, status, responseContent);

        // 设置响应头
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, responseContent.readableBytes());
        response.headers().set(HttpHeaderNames.SERVER, "Netty/4.1");
        response.headers().set(HttpHeaderNames.DATE, new Date());

        // 处理 Keep-Alive
        boolean keepAlive = HttpUtil.isKeepAlive(request);
        if (keepAlive) {
            response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        }

        // 发送响应
        ChannelFuture future = ctx.writeAndFlush(response);

        if (!keepAlive) {
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.err.println("HTTP Server exception: " + cause.getMessage());
        ctx.close();
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108

// WebSocket 服务器实现
public class WebSocketServer {
    private final int port;

    public WebSocketServer(int port) {
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new WebSocketServerInitializer());

            ChannelFuture f = b.bind(port).sync();
            System.out.println("WebSocket Server started on ws://localhost:" + port + "/websocket");

            f.channel().closeFuture().sync();

        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        int port = args.length > 0 ? Integer.parseInt(args[0]) : 8080;
        new WebSocketServer(port).start();
    }
}

// WebSocket 服务器初始化器
class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    public void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();

        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpObjectAggregator(65536));
        pipeline.addLast(new WebSocketServerProtocolHandler("/websocket", null, true));
        pipeline.addLast(new WebSocketServerHandler());
    }
}

// WebSocket 处理器
class WebSocketServerHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
    private static final Set<Channel> channels = ConcurrentHashMap.newKeySet();

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        channels.add(ctx.channel());
        System.out.println("WebSocket client connected: " + ctx.channel().remoteAddress());
        System.out.println("Total clients: " + channels.size());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        channels.remove(ctx.channel());
        System.out.println("WebSocket client disconnected: " + ctx.channel().remoteAddress());
        System.out.println("Total clients: " + channels.size());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (frame instanceof TextWebSocketFrame) {
            // 处理文本消息
            String text = ((TextWebSocketFrame) frame).text();
            System.out.println("Received: " + text);

            // 广播消息给所有客户端
            String response = "Server Echo: " + text + " (Clients: " + channels.size() + ")";
            TextWebSocketFrame responseFrame = new TextWebSocketFrame(response);

            // 广播给所有连接的客户端
            for (Channel channel : channels) {
                if (channel.isActive()) {
                    channel.writeAndFlush(responseFrame.retainedDuplicate());
                }
            }

        } else if (frame instanceof BinaryWebSocketFrame) {
            // 处理二进制消息
            ByteBuf content = frame.content();
            System.out.println("Received binary data: " + content.readableBytes() + " bytes");

            // 回显二进制数据
            ctx.writeAndFlush(new BinaryWebSocketFrame(content.retain()));

        } else if (frame instanceof PingWebSocketFrame) {
            // 处理 Ping 帧
            ctx.writeAndFlush(new PongWebSocketFrame(frame.content().retain()));

        } else if (frame instanceof CloseWebSocketFrame) {
            // 处理关闭帧
            ctx.close();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.err.println("WebSocket exception: " + cause.getMessage());
        ctx.close();
    }
}