Netty简单的入门代码示例

@Slf4j
public class NettyServer{
    //1.创建线程组  bossGroup：连接线程   workGroup：工作线程
    private final NioEventLoopGroup bossGroup = new NioEventLoopGroup();
    private final NioEventLoopGroup workerGroup = new NioEventLoopGroup();
    public void serverStart() throws InterruptedException {
        try{
            // 服务端启动类
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 传入两个线程组
            bootstrap.group(bossGroup, workerGroup)
                    // 指定Channel 和NIO一样是采用Channel通道的方式通信 所以需要指定服务端通道
                    .channel(NioServerSocketChannel.class)
                    //使用指定的端口设置套接字地址
                    .localAddress(new InetSocketAddress(11111))
                    //服务端可连接队列数,对应TCP/IP协议listen函数中backlog参数
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    //设置数据处理器
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel channel) throws Exception {
                            // 在管道中 添加数据处理类
                            channel.pipeline().addLast(new NettyServerTestHandler());
                        }
                    });
            // 同步等待成功
            ChannelFuture future = bootstrap.bind().sync();
            if (future.isSuccess()) {
                log.info("启动 Netty Server 成功");
            }
            //等待服务端监听端口关闭 链路关闭后main函数才会结束
            future.channel().closeFuture().sync();
        }finally {
            // 优雅的关闭 释放资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        new NettyServer().serverStart();
    }
}

channel.pipeline() : 这就是管道
new NettyServerTestHandler() : 这个就是管道里的一个数据处理类
// 数据处理类可以有多个

@Slf4j
public class NettyServerTestHandler extends ChannelInboundHandlerAdapter {
    // 读取信息调用
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 和NIO一样有缓冲区 ByteBuf就是对ByteBuffer做了一层封装
        ByteBuf msg1 = (ByteBuf) msg;
        log.info("客户端信息：" + msg1.toString(CharsetUtil.UTF_8));
    }
    // 连接事件 连接成功调用
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        SocketAddress socketAddress = ctx.channel().remoteAddress();
        log.info(socketAddress + " 已连接");
        // 发送数据
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Client", CharsetUtil.UTF_8));
    }
    // 断开连接调用
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info(ctx.channel().remoteAddress() + " 已断开连接");
    }
    // 读取信息完成事件  信息读取完成后调用
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    }
    // 异常处理  发生异常调用
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 异常后 关闭与客户端连接
        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();
            // 客户端不需要处理连接 所以一个线程组就够了
            bootstrap.group(group)
                    // 连接通道
                    .channel(NioSocketChannel.class)
                    .remoteAddress(host, port)
                    .option(ChannelOption.TCP_NODELAY, true)
                    // 数据处理
                    .handler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel channel) throws Exception {
                            channel.pipeline().addLast(new NettyClientTestHandler());
                        }
                    });
            ChannelFuture future = bootstrap.connect();
            //客户端断线重连逻辑
            future.addListener((ChannelFutureListener) future1 -> {
                if (future1.isSuccess()) {
                    log.info("连接Netty服务端成功");
                } else {
                    log.info("连接失败，进行断线重连");
                    future1.channel().eventLoop().schedule(() -> {
                        try {
                            start();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }, 20, TimeUnit.SECONDS);
                }
            });
            future.channel().closeFuture().sync();
        }catch (Exception e){
            log.info("服务端异常");
        }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("服务端信息：" + msg1.toString(CharsetUtil.UTF_8));
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 连接上 就给服务端发送数据
        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() + " 已断开连接");
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}