物联网基于netty入门与线程模型探秘

简述

理解 Netty 的核心组件（Channel、ChannelHandler、ChannelPipeline），掌握其高性能的 Reactor 线程模型，并通过一个完整的 Echo 示例快速入门

源码

netty-sample-00[https://gitee.com/kcnf-webrtc/iot-sample/tree/master/netty/netty-sample-00]

Netty 核心组件

组件	作用
Channel	代表一个网络连接（如 NIO SocketChannel），负责读写数据。
ChannelHandler	处理 I/O 事件或拦截数据（如解码、业务逻辑）。可分为入站（Inbound）和出站（Outbound）。
ChannelPipeline	管理 ChannelHandler 的链表，数据流经 Pipeline 依次被各 Handler 处理。
EventLoopGroup	一组 EventLoop，每个 EventLoop 绑定一个线程，负责处理多个 Channel 的 I/O 事件。
Bootstrap / ServerBootstrap	引导类，配置并启动 Netty 客户端/服务器。

Reactor 线程模型

三种典型 Reactor 模型

模型	描述	Netty 对应配置
单线程	一个 Reactor 线程同时处理连接、读、写。	EventLoopGroup 大小为 1。
多线程	一个 Reactor 线程处理连接，多个工作线程处理读写。	bossGroup 大小为 1，workerGroup 大小为 CPU*2。
主从多线程	多线程 Reactor 处理连接，多线程 Reactor 处理读写（用于更高并发）。	bossGroup 大小 >1，workerGroup 大小 >1。

Netty 默认实现（主从多线程变种）

BossGroup：监听端口，接收新连接，将 SocketChannel 注册到 WorkerGroup

WorkerGroup：负责已连接的读写 I/O 及业务处理（通常业务 Handler 可再异步执行）

// 典型配置
EventLoopGroup bossGroup = new NioEventLoopGroup(1);   // 单线程处理连接
EventLoopGroup workerGroup = new NioEventLoopGroup();  // 默认 CPU 核数 * 2

每个 EventLoop 内部有一个 Selector 和任务队列，一个 EventLoop 服务于多个 Channel，保证同一 Channel 的事件串行执行

添加pom依赖

 <dependencies>
        <!-- Netty 核心依赖 -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

server 代码

package com.jysemel.iot;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
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 java.nio.charset.StandardCharsets;

public class EchoServer {
    public static void main(String[] args) throws InterruptedException {
        // 1. 创建线程组：boss 接受连接，worker 处理 I/O
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)      // 指定 NIO 传输
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 添加编解码器（处理粘包/拆包可用 LengthFieldBasedFrameDecoder，这里简单演示）
                            pipeline.addLast(new StringDecoder(StandardCharsets.UTF_8));
                            pipeline.addLast(new StringEncoder(StandardCharsets.UTF_8));
                            // 添加业务 Handler
                            pipeline.addLast(new EchoServerHandler());
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 128)      // 连接队列大小
                    .childOption(ChannelOption.SO_KEEPALIVE, true); // 长连接

            // 2. 绑定端口并启动
            ChannelFuture future = bootstrap.bind(8181).sync();
            System.out.println("Echo 服务器启动，端口 8181");
            future.channel().closeFuture().sync();  // 等待直到服务器关闭
        } finally {
            // 3. 优雅关闭
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

client 代码

package com.jysemel.iot;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
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 java.nio.charset.StandardCharsets;

public class EchoClient {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new StringDecoder(StandardCharsets.UTF_8));
                            pipeline.addLast(new StringEncoder(StandardCharsets.UTF_8));
                            pipeline.addLast(new EchoClientHandler());
                        }
                    });

            ChannelFuture future = bootstrap.connect("127.0.0.1", 8181).sync();
            future.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

运行与验证

先启动 EchoServer，再启动 EchoClient。
控制台输出：
服务端返回: Hello, Netty!
服务器输出：收到: Hello, Netty!