【Netty】Socket 编程(C/S)：基于Netty的Server、Client示例（详细注释）

水深无声 2022-12-31 02:29 169阅读 0赞

首先引入 Netty 的依赖

<dependency>
         <groupId>io.netty</groupId>
         <artifactId>netty-all</artifactId>
         <version>4.1.36.Final</version>
    </dependency>

## Server ##

public class NettyServer {
        
    
        public static void main(String[] args) throws Exception {
        
            // 创建两个线程组bossGroup和workerGroup, 含有的子线程NioEventLoop的个数默认为cpu核数的两倍
            // bossGroup只是处理连接请求 ,真正的和客户端业务处理，会交给workerGroup完成
            // 正因为bossGroup仅接收客户端连接，不做复杂的逻辑处理，为了尽可能减少资源的占用，取值越小越好
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            
            try {
        
                // 创建服务器端的启动对象
                ServerBootstrap bootstrap = new ServerBootstrap();
                
                // 使用链式编程来配置参数
                // 设置两个线程组
                bootstrap.group(bossGroup, workerGroup)
                        // 使用NioServerSocketChannel作为服务器的通道实现
                        .channel(NioServerSocketChannel.class)
                        // ChannelInitializer用于配置一个新的Channel
                        // 用于向你的Channel当中添加ChannelInboundHandler的实现
                        .childHandler(new ChannelInitializer<SocketChannel>() {
        //创建通道初始化对象，设置初始化参数
    
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
        
                                // 对workerGroup的SocketChannel设置处理器
                                ch.pipeline().addLast(new NettyServerHandler());
                            }
                        })
                        // 初始化服务器连接队列大小，服务端处理客户端连接请求是顺序处理的,所以同一时间只能处理一个客户端连接。
                        // 多个客户端同时来的时候,服务端将不能处理的客户端连接请求放在队列中等待处理
                        // 标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度。
                        // 如果未设置或所设置的值小于1，Java将使用默认值50。
                        .option(ChannelOption.SO_BACKLOG, 128)
                        // 是否启用心跳保活机制。
                        // 在双方TCP套接字建立连接后（即都进入ESTABLISHED状态）,并且在两个小时左右上层没有任何数据传输的情况下，这套机制才会被激活。
                        // childOption是用来给父级ServerChannel之下的Channels设置参数的
                        .childOption(ChannelOption.SO_KEEPALIVE, true);
    
                
                // 启动服务器(并绑定端口)
                // 绑定一个端口并且同步, 生成了一个ChannelFuture异步对象
                // 注：bind是异步操作，sync方法是等待异步操作执行完毕
                ChannelFuture cf = bootstrap.bind(9000).sync();
                
                System.out.println("Netty Server started...");
    
                // 对通道关闭进行监听，closeFuture是异步操作，监听通道关闭
                // 通过sync方法同步等待通道关闭处理完毕，这里会阻塞等待通道关闭完成
                cf.channel().closeFuture().sync();
            } finally {
        
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

/**
     * 自定义Handler需要继承netty规定好的某个HandlerAdapter(规范)
     */
    public class NettyServerHandler extends ChannelInboundHandlerAdapter {
        
    
        /**
         * 读取客户端发送的数据
         *
         * @param ctx 上下文对象, 含有通道channel，管道pipeline
         * @param msg 就是客户端发送的数据
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
            System.out.println("服务器读取线程 " + Thread.currentThread().getName());
    
            // 将 msg 转成一个 ByteBuf，类似 NIO 的 ByteBuffer
            ByteBuf buf = (ByteBuf) msg;
            System.out.println("客户端发送消息是:" + buf.toString(CharsetUtil.UTF_8));
        }
    
        /**
         * 数据读取完毕处理方法
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        
            ByteBuf buf = Unpooled.copiedBuffer("HelloClient".getBytes(CharsetUtil.UTF_8));
            ctx.writeAndFlush(buf);
        }
    
        /**
         * 处理异常, 一般是需要关闭通道
         *
         * @param ctx
         * @param cause
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
            ctx.close();
        }
    }

## Client ##

public class NettyClient {
        
        public static void main(String[] args) throws Exception {
        
            // 客户端需要一个事件循环组
            EventLoopGroup group = new NioEventLoopGroup();
            try {
        
                // 创建客户端启动对象
                // 注意客户端使用的不是ServerBootstrap而是Bootstrap
                Bootstrap bootstrap = new Bootstrap();
                // 设置相关参数
                //设置线程组
                bootstrap.group(group)
                        // 使用NioSocketChannel作为客户端的通道实现
                        .channel(NioSocketChannel.class) 
                        .handler(new ChannelInitializer<SocketChannel>() {
        
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
        
                                // 加入处理器
                                ch.pipeline().addLast(new NettyClientHandler());
                            }
                        });
    
                // 启动客户端去连接服务器端
                ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync();
                System.out.println("Netty Server started...");
                
                // 对通道关闭进行监听
                cf.channel().closeFuture().sync();
            } finally {
        
                group.shutdownGracefully();
            }
        }
    }

public class NettyClientHandler extends ChannelInboundHandlerAdapter {
        
    
        /**
         * 当客户端连接服务器完成就会触发该方法
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
            ByteBuf buf = Unpooled.copiedBuffer("HelloServer".getBytes(CharsetUtil.UTF_8));
            ctx.writeAndFlush(buf);
        }
    
        //当通道有读取事件时会触发，即服务端发送数据给客户端
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
            ByteBuf buf = (ByteBuf) msg;
            System.out.println("服务端的地址： " + ctx.channel().remoteAddress());
            System.out.println("收到服务端的消息:" + buf.toString(CharsetUtil.UTF_8));
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
            cause.printStackTrace();
            ctx.close();
        }

先后启动 Server 和 Client，结果如下：

![在这里插入图片描述][20201230164633981.png_pic_center]  
![在这里插入图片描述][20201230164650573.png_pic_center]

[20201230164633981.png_pic_center]: /images/20221120/cbcaaa872871492ea95182ecd98ed084.png
[20201230164650573.png_pic_center]: /images/20221120/b18635f82a0745b7bee7fdb063797369.png