netty编程实战01-创建一个tcp服务端程序

<?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.wayn.nettyproject</groupId>
    <artifactId>netty-project</artifactId>
    <version>1.0-SNAPSHOT</version>
    <name>netty-project</name>
    <!-- FIXME change it to the project's website -->
    <url>http://www.example.com</url>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>
    <dependencies>
        <!--netty依赖-->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId> <!-- Use 'netty-all' for 4.0 or above -->
            <version>4.1.72.Final</version>
        </dependency>
        <!--lombok 依赖-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.22</version>
        </dependency>
        <!--##### log4j2 start  ####-->
        <!--log4j2核心包-->
        <!-- https://mvnrepository.com/artifact/org.apache.logging.log4j/log4j-core -->
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.17.1</version>
        </dependency>
        <!--用于与slf4j保持桥接（里面自动依赖了slf4j-api）-->
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
            <version>2.17.1</version>
        </dependency>
        <!-- 异步日志，需要加入disruptor依赖 -->
        <dependency>
            <groupId>com.lmax</groupId>
            <artifactId>disruptor</artifactId>
            <version>3.4.4</version>
        </dependency>
        <!--##### log4j2 end  ####-->
        <!-- fastjson -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.79</version>
        </dependency>
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.7.18</version>
        </dependency>
    </dependencies>
    <build>
    </build>
</project>

/**
 * tcp服务端
 */
@Slf4j
public class NettyServer {
    // 绑定端口
    private static final int port = 99;
    // 维护channel列表，将用户信息和channel做绑定，一个channel代表一个tcp连接
    private static final Map<String, Object> channelMap = new HashMap<>();
    public static void main(String[] args) throws InterruptedException {
        // 创建服务端启动器
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        // 指定连接接收线程，处理所有用户连接，转给工作线程
        NioEventLoopGroup boss = null;
        // 指定工作线程，处理用户连接
        NioEventLoopGroup worker = null;
        try {
            boss = new NioEventLoopGroup();
            worker = new NioEventLoopGroup();
            serverBootstrap.group(boss, worker); // 绑定线程组
            serverBootstrap.channel(NioServerSocketChannel.class);  // 底层使用nio处理
            serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
                protected void initChannel(NioSocketChannel ch) {
                    // 创建一个心跳检测的handle，服务端创建连接后，5秒内没有收到客户端的心跳就会触发IdleState.READER_IDLE
                    IdleStateHandler idleStateHandler = new IdleStateHandler(5, 0, 0) {
                        // 5秒内没有收到客户端的心跳会触发userEventTriggered方法
                        @Override
                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                            if (evt instanceof IdleStateEvent) {
                                IdleStateEvent e = (IdleStateEvent) evt;
                                // 检测是读状态就发送pong消息
                                if (e.state() == IdleState.READER_IDLE) {
                                    ByteBuf buffer = Unpooled.buffer();
                                    buffer.writeBytes("pong".getBytes(StandardCharsets.UTF_8));
                                    ctx.channel().writeAndFlush(buffer).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                                    String id = ctx.channel().id().asLongText();
                                    // 删除channelMap中客户端连接
                                    channelMap.forEach((s, o) -> {
                                        if (s.equals(id)) {
                                            channelMap.remove(s);
                                        }
                                    });
                                }
                            }
                        }
                        // 发生异常时，关闭连接
                        @Override
                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                            log.error(cause.getMessage(), cause);
                            String id = ctx.channel().id().asLongText();
                            channelMap.remove(id);
                            ctx.close();
                        }
                    };
                    // 通过对象实例，而不是new创建，避免大量客户端连接导致浪费空间
                    ch.pipeline().addLast("idleStateHandler", idleStateHandler);
                    // 换行符解码器，指定包与包之间的分隔符是"\n"和"\r\n",避免拆包粘包
                    ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                    // netty日志记录，打印包信息
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO));
                    // 入站字符解码器
                    ch.pipeline().addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
                    // 自定义解码器，实现自定义业务逻辑，使用SimpleChannelInboundHandler可以避免ByteBuf的回收问题
                    ch.pipeline().addLast("MyHandle", new SimpleChannelInboundHandler<String>() {
                        // 连接第一次创建会调用这个方法
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            String id = ctx.channel().id().asLongText();
                            log.info("客户端建立连接：{}", id);
                            channelMap.put(id, ctx.channel());
                            ctx.channel().writeAndFlush("this is come from server!");
                            super.channelActive(ctx);
                        }
                        // 获取来自客户端的数据
                        @Override
                        protected void channelRead0(ChannelHandlerContext ctx, String s) {
                            log.info("收到客户端消息：{}", s);
                            ctx.channel().writeAndFlush("server reply\r\n");
                        }
                    });
                    // 出站字符编码器
                    ch.pipeline().addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
                }
            });
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync().addListener(future -> {
                if (future.isSuccess()) {
                    log.info("server start up on {}", port);
                }
            });
            channelFuture.channel().closeFuture().sync();
        } finally {
            // 不要忘了在finally中关闭线程组
            if (boss != null) {
                boss.shutdownGracefully();
            }
            if (worker != null) {
                worker.shutdownGracefully();
            }
        }
    }
}