@Slf4j
public class NettyServer{
    //1.創(chuàng)建線程組  bossGroup：連接線程   workGroup：工作線程
    private final NioEventLoopGroup bossGroup = new NioEventLoopGroup();
    private final NioEventLoopGroup workerGroup = new NioEventLoopGroup();
    public void serverStart() throws InterruptedException {
        try{
            // 服務(wù)端啟動(dòng)類(lèi)
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 傳入兩個(gè)線程組
            bootstrap.group(bossGroup, workerGroup)
                    // 指定Channel 和NIO一樣是采用Channel通道的方式通信 所以需要指定服務(wù)端通道
                    .channel(NioServerSocketChannel.class)
                    //使用指定的端口設(shè)置套接字地址
                    .localAddress(new InetSocketAddress(11111))
                    //服務(wù)端可連接隊(duì)列數(shù),對(duì)應(yīng)TCP/IP協(xié)議listen函數(shù)中backlog參數(shù)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    //設(shè)置數(shù)據(jù)處理器
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel channel) throws Exception {
                            // 在管道中 添加數(shù)據(jù)處理類(lèi)
                            channel.pipeline().addLast(new NettyServerTestHandler());
                        }
                    });
            // 同步等待成功
            ChannelFuture future = bootstrap.bind().sync();
            if (future.isSuccess()) {
                log.info("啟動(dòng) Netty Server 成功");
            }
            //等待服務(wù)端監(jiān)聽(tīng)端口關(guān)閉 鏈路關(guān)閉后main函數(shù)才會(huì)結(jié)束
            future.channel().closeFuture().sync();
        }finally {
            // 優(yōu)雅的關(guān)閉 釋放資源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        new NettyServer().serverStart();
    }
}

channel.pipeline() : 這就是管道
new NettyServerTestHandler() : 這個(gè)就是管道里的一個(gè)數(shù)據(jù)處理類(lèi)
// 數(shù)據(jù)處理類(lèi)可以有多個(gè)

@Slf4j
public class NettyServerTestHandler extends ChannelInboundHandlerAdapter {
    // 讀取信息調(diào)用
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 和NIO一樣有緩沖區(qū) ByteBuf就是對(duì)ByteBuffer做了一層封裝
        ByteBuf msg1 = (ByteBuf) msg;
        log.info("客戶(hù)端信息：" + msg1.toString(CharsetUtil.UTF_8));
    }
    // 連接事件 連接成功調(diào)用
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        SocketAddress socketAddress = ctx.channel().remoteAddress();
        log.info(socketAddress + " 已連接");
        // 發(fā)送數(shù)據(jù)
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Client", CharsetUtil.UTF_8));
    }
    // 斷開(kāi)連接調(diào)用
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info(ctx.channel().remoteAddress() + " 已斷開(kāi)連接");
    }
    // 讀取信息完成事件  信息讀取完成后調(diào)用
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    }
    // 異常處理  發(fā)生異常調(diào)用
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 異常后 關(guān)閉與客戶(hù)端連接
        ctx.close();
    }
}

@Slf4j
public class NettyClient  {
    private EventLoopGroup group = new NioEventLoopGroup();
    private int port=11111;
    private String host="127.0.0.1";
    public void start() throws InterruptedException {
        try{
            Bootstrap bootstrap = new Bootstrap();
            // 客戶(hù)端不需要處理連接 所以一個(gè)線程組就夠了
            bootstrap.group(group)
                    // 連接通道
                    .channel(NioSocketChannel.class)
                    .remoteAddress(host, port)
                    .option(ChannelOption.TCP_NODELAY, true)
                    // 數(shù)據(jù)處理
                    .handler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel channel) throws Exception {
                            channel.pipeline().addLast(new NettyClientTestHandler());
                        }
                    });
            ChannelFuture future = bootstrap.connect();
            //客戶(hù)端斷線重連邏輯
            future.addListener((ChannelFutureListener) future1 -> {
                if (future1.isSuccess()) {
                    log.info("連接Netty服務(wù)端成功");
                } else {
                    log.info("連接失敗，進(jìn)行斷線重連");
                    future1.channel().eventLoop().schedule(() -> {
                        try {
                            start();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }, 20, TimeUnit.SECONDS);
                }
            });
            future.channel().closeFuture().sync();
        }catch (Exception e){
            log.info("服務(wù)端異常");
        }finally {
            group.shutdownGracefully();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        new NettyClient().start();
    }
}

@Slf4j
public class NettyClientTestHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf msg1 = (ByteBuf) msg;
        log.info("服務(wù)端信息：" + msg1.toString(CharsetUtil.UTF_8));
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 連接上 就給服務(wù)端發(fā)送數(shù)據(jù)
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Server", CharsetUtil.UTF_8));
        SocketAddress socketAddress = ctx.channel().remoteAddress();
        log.info(socketAddress + " 已連接");
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info(ctx.channel().remoteAddress() + " 已斷開(kāi)連接");
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}