Netty-EchoServer回显服务器的案例

小鱼儿 2023-06-16 10:38 30阅读 0赞

### 文章目录 ###

*  EchoServer回显服务器的实践案例
 *   *  服务器端
     *   *  功能
         *   *  服务器端需要掌握的知识
         *  code
         *   *  NettyEchoServer
             *  Handler处理器
             *   *  回显服务器处理的逻辑
         *  客户端
         *   *  NettyEchoClient
             *  NettyEchoClientHandler处理器

# EchoServer回显服务器的实践案例 #

## 服务器端 ##

### 功能 ###

> 从服务器端读取客户端输入的数据，然后将数据直接回显示在Console控制台

#### 服务器端需要掌握的知识 ####

*  服务器端ServerBootstrap的装配和使用
 *  NettyEchoServerHandler入站处理器的入站方法编写
 *  Netty的ByteBuf缓冲区的读取，写入，以及ByteBuf的引用计数的查看

### code ###

#### NettyEchoServer ####

package com.wangyg.NettyDemos.echoServer;
    
    import com.wangyg.netty.basic.NettyDemoConfig;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.PooledByteBufAllocator;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import util.Logger;
    
    
    public class NettyEchoServer {
        private final int serverPort;
        ServerBootstrap b = new ServerBootstrap();
    
        public NettyEchoServer(int port) {
            this.serverPort = port;
        }
    
        public void runServer() {
            //创建reactor 线程组
            EventLoopGroup bossLoopGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerLoopGroup = new NioEventLoopGroup();
    
            try {
                //1 设置reactor 线程组
                b.group(bossLoopGroup, workerLoopGroup);
                //2 设置nio类型的channel
                b.channel(NioServerSocketChannel.class);
                //3 设置监听端口
                b.localAddress(serverPort);
                //4 设置通道的参数
                b.option(ChannelOption.SO_KEEPALIVE, true);
                b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
                b.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    
                //5 装配子通道流水线
                b.childHandler(new ChannelInitializer<SocketChannel>() {
                    //有连接到达时会创建一个channel
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // pipeline管理子通道channel中的Handler
                        // 向子channel流水线添加一个handler处理器
                        ch.pipeline().addLast(NettyEchoServerHandler.INSTANCE);
                    }
                });
                // 6 开始绑定server
                // 通过调用sync同步方法阻塞直到绑定成功
                ChannelFuture channelFuture = b.bind().sync();
                Logger.info(" 服务器启动成功，监听端口: " +
                        channelFuture.channel().localAddress());
    
                // 7 等待通道关闭的异步任务结束
                // 服务监听通道会一直等待通道关闭的异步任务结束
                ChannelFuture closeFuture = channelFuture.channel().closeFuture();
                closeFuture.sync();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 8 优雅关闭EventLoopGroup，
                // 释放掉所有资源包括创建的线程
                workerLoopGroup.shutdownGracefully();
                bossLoopGroup.shutdownGracefully();
            }
    
        }
    
        public static void main(String[] args) throws InterruptedException {
            int port = 8888;
            new NettyEchoServer(port).runServer();
        }
    }

#### Handler处理器 ####

##### 回显服务器处理的逻辑 #####

*  从channelRead方法的msg参数
 *  调用ctx.channel().writeAndFlush()把数据写回客户端

package com.wangyg.NettyDemos.echoServer;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import org.jboss.netty.channel.ChannelHandler;
    import util.Logger;
    
    /**
     * 标注一个channel handler可以被多个channel安全地共享
     *
     * 会先服务器处理器的逻辑分为两步
     *:
     * 第一步： 从channelRead方法的msg参数
     * 第二步： 调用ctx.channel().writeAndFlush()把数据写回到客户端
     *
     */
    @ChannelHandler.Sharable //标注一个channel handler可以被多个channel安全的共享：共享就是多个通道的流水线可以加入同一个Handler业务处理器实例
    //而这种操作，Netty默认是不允许的, 很多场景都需要Handler业务处理器实例共享，例如： 一个服务器处理十万以上的通道，如果一个通道都新建很多
    //重复的Handler实例，就会浪费很多宝贵的空间，降低服务器的性能
    public class NettyEchoServerHandler  extends ChannelInboundHandlerAdapter {
        public static final NettyEchoServerHandler INSTANCE = new NettyEchoServerHandler();
        /**
         * msg 的类型是由流水线的上一站决定的，
         *
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf in = (ByteBuf) msg;
            Logger.info("msg type: " + (in.hasArray()?"堆内存":"直接内存"));
    
            int len = in.readableBytes();
            byte[] arr = new byte[len];
            in.getBytes(0, arr);
            Logger.info("server received: " + new String(arr, "UTF-8"));
    
            /**
             * 回写给客户端:
             * 直接复用前面的msg实例即可
             * 上一步调用了getBytes，不影响ByteBuf的数据指针，因此可以继续使用,所以此处调用了ctx.writeAndFlush
             * 将msg写回到客户端
             */
            //写回数据，异步任务
            Logger.info("写回前，msg.refCnt:" + ((ByteBuf) msg).refCnt());
            ChannelFuture f = ctx.writeAndFlush(msg); //调用wrteAndFlush
            f.addListener((ChannelFuture futureListener) -> {
                Logger.info("写回后，msg.refCnt:" + ((ByteBuf) msg).refCnt());
            });
        }
    }

@ChannelHandler.Sharable

> 标注一个Handler实例可以被多个通道安全共享, 共享就是多个通道的流水线可以加入同一个Handler业务处理器实例

### 客户端 ###

客户端的实践，目标为掌握以下知识：

*  客户端的Bootstrap的装配和使用
 *  客户端NettyEchoClientHandler入站处理器中，接受回写的数据，并释放内存
 *  多种方式用于释放ByteBuf,包括： 自动释放，手动释放

#### NettyEchoClient ####

package com.wangyg.NettyDemos.echoServer;
    
    import com.wangyg.netty.basic.NettyDemoConfig;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.PooledByteBufAllocator;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import util.Dateutil;
    import util.Logger;
    import util.Print;
    
    import java.io.UnsupportedEncodingException;
    import java.util.Scanner;
    
    public class NettyEchoClient {
        private int serverPort;
        private String serverIp;
    
        Bootstrap b = new Bootstrap();
    
        /**
         * 构造函数
         *
         * @param ip
         * @param port
         */
        public NettyEchoClient(String ip, int port) {
            this.serverIp = ip;
            this.serverPort = port;
        }
    
        public void runClient() {
            //创建reactor线程组
            EventLoopGroup workerLoopGroup = new NioEventLoopGroup();
    
            try {
                //设置reactor线程组
                b.group(workerLoopGroup);
                //设置Nio类型的channel
                b.channel(NioSocketChannel.class);
                //设置监听端口
                b.remoteAddress(serverIp, serverPort);
                //设置通道的参数
                b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    
                //装配子通道流水线
                b.handler(new ChannelInitializer<SocketChannel>() {
                    //有链接到达时会创建一个channel
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        //pipeline管理子通道channel中的handler
                        //向子channel流水线添加一个handler处理器
                        ch.pipeline().addLast(NettyEchoClientHandler.INSTANCE);
                    }
                });
                ChannelFuture f = b.connect();
                //添加监听器
                f.addListener((ChannelFuture futureListener) -> {
                    if (futureListener.isSuccess()) {
                        Logger.info("EchoClient客户端连接成功!");
    
                    } else {
                        Logger.info("EchoClient客户端连接失败!");
                    }
                });
    
                // 阻塞,直到连接完成
                f.sync();
                Channel channel = f.channel();
    
                Scanner scanner = new Scanner(System.in);
                Print.tcfo("请输入发送内容:");
    
                while (scanner.hasNext()) {
                    //获取输入的内容
                    String next = scanner.next();
                    byte[] bytes = (Dateutil.getNow() + " >>" + next).getBytes("UTF-8");
                    //发送ByteBuf
                    ByteBuf buffer = channel.alloc().buffer();
                    buffer.writeBytes(bytes);
                    channel.writeAndFlush(buffer);
                    Print.tcfo("请输入发送内容:");
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 优雅关闭EventLoopGroup，
                // 释放掉所有资源包括创建的线程
                workerLoopGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) {
            int port = 8888;
            String ip = "127.0.0.1";
            new NettyEchoClient(ip, port).runClient();
        }
    }

#### NettyEchoClientHandler处理器 ####

package com.wangyg.NettyDemos.echoServer;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import util.Logger;
    
    public class NettyEchoClientHandler  extends ChannelInboundHandlerAdapter {
        public static  final  NettyEchoClientHandler INSTANCE = new NettyEchoClientHandler();
    
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf in = (ByteBuf) msg;
            int len = in.readableBytes();
            byte[] arr = new byte[len];
            in.getBytes(0, arr);
            Logger.info("client received: " + new String(arr, "UTF-8"));
            in.release();
        }
    }

效果图  
![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-18PIOSL5-1574692907847)(F298392E385E4E6FA9B088DE3C9DA09F)\]][img-18PIOSL5-1574692907847_F298392E385E4E6FA9B088DE3C9DA09F]

[img-18PIOSL5-1574692907847_F298392E385E4E6FA9B088DE3C9DA09F]: https://img-blog.csdnimg.cn/20191125224203540.png