使用Netty编写一个Echo程序

布满荆棘的人生 2022-09-11 14:17 158阅读 0赞

Echo服务程序：实现客户端发送消息给服务器，服务器将收到的消息返回给客户端。Echo服务程序可以用于程序调试和检测。

package com.tech.netty.echo;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoServer {
    
        private int port;
    
        public EchoServer(int port) {
            this.port = port;
        }
    
        /**
         * 启动流程
         * @throws InterruptedException
         */
        public void run() throws InterruptedException {
            //配置服务端线程组
            NioEventLoopGroup bossGroup = new NioEventLoopGroup();
            NioEventLoopGroup workerGroup = new NioEventLoopGroup();
            try{
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup,workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new EchoServerHandler());
                            }
                        });
                System.out.println("Echo 服务器启动ing");
                //绑定端口 同步等待成功
                ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
                //等待服务端监听端口关闭
                channelFuture.channel().closeFuture().sync();
            }finally {
                //优雅退出释放线程池
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            int port=8080;
            if(args.length>0){
                port=Integer.parseInt(args[0]);
            }
            new EchoServer(port).run();
        }
    }

package com.tech.netty.echo;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoServerHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf byteBuf = (ByteBuf) msg;
            System.out.println("服务端收到数据："+byteBuf.toString(CharsetUtil.UTF_8));
            ctx.writeAndFlush(byteBuf);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            System.out.println("EchoServerHandler channelReadComplete");
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

package com.tech.netty.echo;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoClient {
        private int port;
        private String host;
    
        public EchoClient(int port, String host) {
            this.port = port;
            this.host = host;
        }
    
        private void start() throws InterruptedException {
            NioEventLoopGroup group = new NioEventLoopGroup();
            try{
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new EchoClientHandler());
                            }
                        });
                //连接到服务端，connect是异步连接，在调用同步等待sync，等待连接成功
                ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
                //阻塞直到客户端通道关闭
                channelFuture.channel().closeFuture().sync();
            }finally {
                //优雅退出，释放NIO线程组
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new EchoClient(8080,"127.0.0.1").start();;
        }
    }

package com.tech.netty.echo;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.util.CharsetUtil;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            System.out.println("Client received："+msg.toString(CharsetUtil.UTF_8));
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("Active");
            ctx.writeAndFlush(Unpooled.copiedBuffer("调试消息",CharsetUtil.UTF_8));
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            System.out.println("EchoClientHandler channelReadComplete");
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }