Netty简单入门

Posted 2021-04-30 小楼杂记

Netty是一个基于异步事件驱动的网络应用系统框架。支持快速地开发可维护的高性能的面向协议的服务器和客户端，业界有很多实用。如：分布式Dubbo采用Netty作为通信组件， Hadoop 的高性能通信和序列化组件 Avro 默认采用Netty进行跨节点通信。

从传统IO到Netty编程

BIO

基于Java对网络进行编程的时候，往往最开始采用的是基于传统BIO的方式进行Socket编程。
例如写一个Http服务：

1. 创建一个ServerSocket,监听并绑定一个端口。

2. 客户端进行请求这个端口。

3. 服务器采用Accept方法，阻塞等待客户端的连接。

4. 获取客户端连接后，启动一个线程处理连接。

1. 读Socket获取，得到字节流。

2. 解码协议，得到Http请求对象。

3. 处理Http请求，得到一个结果，封装成HttpResponse对象。

4. 解码协议，将响应对象序列化成字节流。

5. 将字节流写入Socket，发送到客户端。

5. 循环执行第三步

上面的网络编程是基于传统IO进行编写，这样实现的服务器程序，对一个客户端连接都会创建一个线程进行请求，即“一个请求，一个线程”，当然可以采用线程池的方式进行优化处理，当还是存在一些问题，比如：

• 在任何时候都有可能导致大量的线程处于休眠状态，只是等待输入或输出，会造成资源浪费；

• 需要为一个线程的调用栈分配内存，大小在64KB到1MB之间，具体取决于操作系统。

• 即是内存足以支撑创建足够多的线程，当时不同线程之前的上下文切换也会造成系统资源的开销。

BIO代码示例：

 1        final ServerSocket serverSocket = new ServerSocket(port);
 2        Socket clientSocket = serverSocket.accept();
 3        BufferedReader in = new BufferedReader(
 4                new InputStreamReader(clientSocket.getInputStream()));
 5        PrintWriter out =
 6                new PrintWriter(clientSocket.getOutputStream(), true);
 7        String request, response;
 8        while ((request = in.readLine()) != null) {
 9            if ("Done".equals(request)) {
10                break;
11            }
12            // 创建一个新线程处理请求
13            response = processRequest(request);
14            out.println(response);
15        }

NIO

NIO的全称是NoneBlocking IO，非阻塞IO，是相对于BIO（Blocking IO）而阻塞而言的，那传统IO被称为阻塞IO，是因为在等待Accpet的时候，线程是阻塞的，在读或写消息的时候，线程是阻塞的。而NIO不一样，它采用事件机制，它可以在一个线程中进行Accept、读写操作、处理请求逻辑。如果没有事件处理，则会进行休眠。
NIO本质是采用了操作系统的IO多路复用技术：一个线程，通过记录I/O流的状态来同时管理多个I/O，可以提高服务器的吞吐能力。其实IO多路复用的实现有多种方式：select、poll、epoll。

   NIO模型

相关概念：

• Selector：操作系统NIO的选择器功能，可以使用操作系统的事件API注册一组非阻塞的Socket。

• Channel：Channel代表一个实体的开放连接，它是输入或输出的载体，实体可以是一个文件、一个硬件设备、一个网络套接字。

Java NIO简易实例：

 1 public void server(int port) throws IOException {
 2        ServerSocketChannel socketChannel = ServerSocketChannel.open();
 3        socketChannel.configureBlocking(false);
 4        ServerSocket serverSocket = socketChannel.socket();
 5        InetSocketAddress address = new InetSocketAddress(port);
 6        serverSocket.bind(address);
 7        Selector selector = Selector.open();
 8        // 将ServerSocket注册到Selector以接受连接
 9        socketChannel.register(selector, SelectionKey.OP_ACCEPT);
10        for (;;) {
11            try {
12                //  等待需要处理的新事件;阻塞将一直持续到下一个传入事件
13                selector.select();
14            } catch (IOException ex) {
15                ex.printStackTrace();
16                break;
17            }
18            // 获取所有接受事件的SelectionKey的实例
19            Set<SelectionKey> readKeys = selector.selectedKeys();
20            Iterator<SelectionKey> iterator = readKeys.iterator();
21
22            while (iterator.hasNext()) {
23                SelectionKey key = iterator.next();
24                iterator.remove();
25                try {
26                    if (key.isAcceptable()) {
27                        ServerSocketChannel server = (ServerSocketChannel) key.channel();
28                        SocketChannel client = server.accept();
29                        client.configureBlocking(false);
30                        client.register(selector, SelectionKey.OP_WRITE | SelectionKey.OP_READ));
31                        if (key.isReadable()) {
32                             SocketChannel channel = (SocketChannel) key.channel();
33                            // do somethin read
34                        }
35                        if (key.isWritable()) {
36                            // do somethin write
37                        }
38                        client.close();
39                    }
40                } catch (IOException ex) {
41                    key.cancel();
42                    key.channel().close();
43                }
44            }
45        }
46    }

Netty

采用上面的NIO确实能够提升应用程序的效率，但是JDK提供的NIO曾经出现过epoll死循环bug，JDK提供的类库非常复杂，还出现过epoll死循环的bug，想用Java NIO写出高可靠性的程序难度很大，于是Netty产生了。
Netty的优点：

1. API使用简单，更容易上手，开发门槛低

2. 功能强大，预置了多种编解码功能，支持多种主流协议

3. 定制能力高，可以通过ChannelHandler对通信框架进行灵活地拓展

4. 高性能，与目前多种NIO主流框架相比，Netty综合性能最高

5. 高稳定性，解决了JDK NIO的BUG

那相比Java NIO有哪些不同呢？Netty在Java NIO的基础上，提供了更高层的抽象，带来高性能、高可靠的体验。

Netty Demo

上面介绍了这么多，下面开始基于Netty进行开发一个简单的Demo。Echo 服务器，即将服务器客户端的输入进行输出到客户端。

创建业务逻辑处理类

Netty为了处理数据采用了责任链模式处理输入与输出数据，每个handler都要继承实现ChannelHandler接口。ChannelHandler，它是一个接口族的父接口，它的实现负责接收并响应事件通知。在 Netty 应用程序中，所有的数据处理逻辑都包含在这些核心抽象的实现中。而Handler分成InboundHanler（入站）与OutboundHandler（出站），然后采用ChannelPipeline将handler按照顺序组成处理流水线。

因为Echo 服务器会响应传入的消息，所以它需要实现 ChannelInboundHandler 接口， 用来定义响应入站事件的方法 。具体业务逻辑类可以通过继承ChannelInboundHandlerAdapter进行轻松实现。

Netty handle流水线

 [e-mail]
 2public class EchoServerHandler extends ChannelInboundHandlerAdapter {
 3
 4    // 开始读取数据
 5    @Override
 6    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 7        ByteBuf in = (ByteBuf) msg;
 8        System.out.println("Server received: " + in.toString(CharsetUtil.UTF_8));
 9        ctx.write(in);
10    }
11
12    // 读取完客户端数据
13    @Override
14    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
15        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
16                .addListener(ChannelFutureListener.CLOSE);
17    }
18
19    // 在处理过程中引发异常是被调用
20    @Override
21    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
22        cause.printStackTrace();
23        ctx.close();
24    }
25}

引导服务类

ChannelHandler负责实现具体的业务逻辑，而引导类需要单独实现。引导类主要有两个功能：

• 绑定到服务器将在其上监听并接受传入连接请求的端口；

• 配置 Channel，以将有关的入站消息通知给 EchoServerHandler 实例。

 1public class EchoServer {
 2    private final int port;
 3
 4    public EchoServer(int port) {
 5        this.port = port;
 6    }
 7
 8    public void start() throws Exception {
 9        final EchoServerHandler serverHandler = new EchoServerHandler();
10        // 创建EventLoopGroup NioEventLoopGroup是一个处理I/O操作的多线程事件循环
11        // 通常称为“boss”，接受传入的连接。
12        EventLoopGroup boss = new NioEventLoopGroup();
13        // 通常称为“worker”，当boss接受连接并注册被接受的连接到worker时，处理被接受连接的流量
14        EventLoopGroup worker = new NioEventLoopGroup();
15
16        try {
17            // server引导类
18            ServerBootstrap b = new ServerBootstrap();
19            b.group(boss, worker)
20                    // 指定所使用的NIO传输Channel
21                    .channel(NioserverSocketChannel.class)
22                    // 使用指定的端口设置Socket地址
23                    .localAddress(new InetSocketAddress(port))
24                    // 添加EchoServerHandler到子Channel的ChannelPipeline
25                    .childHandler(new ChannelInitializer<SocketChannel>() {
26                        @Override
27                        protected void initChannel(SocketChannel ch) throws Exception {
28                            ch.pipeline().addLast(serverHandler);
29                        }
30                    });
31            // 异步绑定服务器；调用sync()方法阻塞等待直到绑定完成
32            ChannelFuture f = b.bind().sync();
33            // 获取Channel的CloseFuture,当且阻塞当前线程直到它完成
34            f.channel().closeFuture().sync();
35        } finally {
36            // 关闭EventLoopGroup 释放所有的资源
37            boss.shutdownGracefully().sync();
38            worker.shutdownGracefully().sync();
39        }
40    }
41}

客户端实现

 [e-mail]
 2public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
 3
 4    // 在到服务器的连接已经建立之后被调用
 5    @Override
 6    public void channelActive(ChannelHandlerContext ctx) throws Exception {
 7       ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks", CharsetUtil.UTF_8));
 8    }
 9
10    // 当从服务器接收到一条消息时被调用
11    @Override
12    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
13        System.out.println("Client received: "+msg.toString(CharsetUtil.UTF_8));
14    }
15
16    // 在处理过程中引发异常是被调用
17    @Override
18    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
19       cause.printStackTrace();
20       ctx.close();
21    }
22}

 1public class EchoClient {
 2    private final String host;
 3    private final int port;
 4
 5    public EchoClient(String host, int port) {
 6        this.host = host;
 7        this.port = port;
 8    }
 9
10    public void start() throws Exception{
11        // 客户端引导类只用一个eventLoopGroup
12        EventLoopGroup group = new NioEventLoopGroup();
13        try {
14            Bootstrap b = new Bootstrap();
15            b.group(group)
16                    .channel(NioSocketChannel.class)
17                    .remoteAddress(new InetSocketAddress(host,port))
18                    .handler(new ChannelInitializer<SocketChannel>() {
19                        @Override
20                        protected void initChannel(SocketChannel ch) throws Exception {
21                            ch.pipeline().addLast(new EchoClientHandler());
22                        }
23                    });
24             // 连接服务端
25            Channel ch = b.connect(host, port).sync().channel();
26            // 控制台输入
27            BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
28
29            for (; ; ) {
30                String line = in.readLine();
31                if (line == null) {
32                    continue;
33                }
34                ch.writeAndFlush(line + "\n");
35            }
36        }finally {
37            group.shutdownGracefully().sync();
38        }
39    }

总结

本文介绍了从Java BIO到NIO，再到使用Netty的简易使用，希望能够对Netty这个高性能的网络框架多一些了解，与掌握NIO的基本原理。