Netty 入门实战

桃扇骨 2022-02-13 09:51 185阅读 0赞

简单介绍Netty的核心概念，这个很新概念就是学习Netty是如何拦截和处理异常，对于刚开始学习netty的读者，利用netty的异常拦截机制来调试程序问题很有帮助。本章还会介绍其他一些核心概念，如服务器和客户端的启动以及分离通道的处理程序。本章学习一些基础以便后面章节的深入学习。本章中将编写一个基于netty的服务器和客户端来互相通信，我们首先来设置netty的开发环境。

## 编写一个应答服务器 ##

写一个Netty服务器主要由两部分组成：  
• 配置服务器功能，如线程、端口  
• 实现服务器处理程序，它包含业务逻辑，决定当有一个请求连接或接收数据时该做什么

**启动服务器**  
通过创建ServerBootstrap对象来启动服务器，然后配置这个对象的相关选项，如端口、线程模式、事件循环，并且添加逻辑处理程序用来处理业务逻辑。

package cn.org.july.netty.one.demo.server;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /**
     * Created with IntelliJ IDEA.
     * User:  wanghongjie
     * Date:  2019/4/26 - 22:43
     * <p>
     * Description:
     */
    public class EchoServer {
        /**
         * 服务端口号
         */
        private final int port;
    
        public EchoServer(int port) {
            this.port = port;
        }
    
        public void start() throws InterruptedException {
            EventLoopGroup executors = new NioEventLoopGroup();
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(executors, group).channel(NioServerSocketChannel.class).localAddress(port)
                        .childHandler(new ChannelInitializer<Channel>() {
                            protected void initChannel(Channel channel) throws Exception {
                                channel.pipeline().addLast(new EchoServerHandler());
                            }
                        });
                ChannelFuture f = serverBootstrap.bind().sync();
                System.out.println(EchoServer.class.getName() + "started and listen on “" + f.channel().localAddress());
                f.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                executors.shutdownGracefully().sync();
            }
        }
    
        public static void main(String[] args) {
            EchoServer echoServer = new EchoServer(5210);
            try {
                echoServer.start();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

**实现服务器业务逻辑**  
Netty使用futures和回调概念，它的设计允许你处理不同的事件类型，更详细的介绍将再后面章节讲述，但是我们可以接收数据。你的channel handler必须继承ChannelInboundHandlerAdapter并且重写channelRead方法，这个方法在任何时候都会被调用来接收数据，在这个例子中接收的是字节。

package cn.org.july.netty.one.demo.server;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    import java.io.UnsupportedEncodingException;
    
    /**
     * Created with IntelliJ IDEA.
     * User:  wanghongjie
     * Date:  2019/4/26 - 22:51
     * <p>
     * Description:
     */
    public class EchoServerHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf byteBuf = (ByteBuf) msg;
            byte[] tempBuf = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(tempBuf);
            System.out.println("Server received: " + new String(tempBuf, "UTF-8"));
            ctx.writeAndFlush(Unpooled.copiedBuffer(tempBuf));
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws UnsupportedEncodingException {
            System.out.println("channelActive:" + ctx.channel().remoteAddress());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

**编写应答程序的客户端**  
服务器写好了，现在来写一个客户端连接服务器。应答程序的客户端包括以下几步：  
• 连接服务器  
• 写数据到服务器  
• 等待接受服务器返回相同的数据  
• 关闭连接  
引导客户端  
引导客户端启动和引导服务器很类似，客户端需同时指定host和port来告诉客户端连接哪个服务器。

package cn.org.july.netty.one.demo.client;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    import java.net.InetSocketAddress;
    
    /**
     * Created with IntelliJ IDEA.
     * User:  wanghongjie
     * Date:  2019/4/26 - 23:08
     * <p>
     * Description:
     */
    public class EchoClient {
        private final int port;
        private final String host;
    
        public EchoClient(String host, int port) {
            this.host = host;
            this.port = port;
        }
    
        public void start() throws InterruptedException {
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group).channel(NioSocketChannel.class).remoteAddress(new InetSocketAddress(host, port))
                        .handler(new ChannelInitializer<SocketChannel>() {
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                socketChannel.pipeline().addLast(new EchoClientHandler());
                            }
                        });
                ChannelFuture f = bootstrap.connect().sync();
                f.channel().closeFuture().sync();
            } catch (Exception e) {
                group.shutdownGracefully().sync();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new EchoClient("localhost", 5210).start();
        }
    
    }

**实现客户端的业务逻辑**

package cn.org.july.netty.one.demo.client;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.util.CharsetUtil;
    import io.netty.util.ReferenceCountUtil;
    
    /**
     * Created with IntelliJ IDEA.
     * User:  wanghongjie
     * Date:  2019/4/27 - 09:42
     * <p>
     * Description:
     */
    @ChannelHandler.Sharable
    public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
    
            System.out.println(byteBuf.toString(CharsetUtil.UTF_8));
            byte[] tempBuf = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(tempBuf);
            System.out.println("netty client send msg is : " + new String(tempBuf, "UTF-8"));
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Netty !",
                    CharsetUtil.UTF_8));
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            super.exceptionCaught(ctx, cause);
        }
    }

在这里使用的是SimpleChannelInboundHandler而不使用ChannelInboundHandlerAdapter，主要原因是ChannelInboundHandlerAdapter在处理完消息后需要负责释放资源。在这里将调用ByteBuf.release()来释放资源。SimpleChannelInboundHandler会在完成channelRead0后释放消息，这是通过Netty处理所有消息的ChannelHandler实现了ReferenceCounted接口达到的。  
为什么在服务器中不使用SimpleChannelInboundHandler呢？因为服务器要返回相同的消息给客户端，在服务器执行完成写操作之前不能释放调用读取到的消息，因为写操作是异步的，一旦写操作完成后，Netty中会自动释放消息。  
![在这里插入图片描述][20190427171403580.gif]

[20190427171403580.gif]: /images/20220213/ac2bc813bc3446b7ba33cf8d1447f11a.png