RPC一-线程模型 · 造舟野渡

RPC系列

概述

作为网络框架，线程模型设计的好坏直接影响其性能。下面我尝试理解netty作为一个prc的底层线程模型。

好几年前，我就听说netty是一个高性能的网络框架，好多框架底层通讯都基于它。先去Google一下netty为什么性能好，得到回复总结起来就是：事件驱动，异步非阻塞。当时还是无法理解，异步非阻塞为什么性能就好，对java来说，请求进来，每个都得在单独的线程栈中处理，异步也只不过换了一个线程处理，多了cpu切换，该走的逻辑都得走，怎么就高性能了？

回答这个问题，我们得先分析以前的玩法。

几个概念

IO操作都有哪些

我们知道，IO就是Input/Output的意思，读写文件就算IO操作。linux有一个概念：一切皆文件，所以普通文件、目录访问、管道、命令、设备、网络等操作都是IO操作。因为计算机硬件的限制，IO操作速度远远赶不上CUP运行速度，所以IO操作会阻塞线程往下执行。

常说的性能指标

响应时间(RT)：指系统对请求作出响应的时间。
吞吐量(Throughput)：单位时间内处理的请求数。好比一条高速同时能上多少车辆。
QPS：每秒查询率，和吞吐量有点像，即最大吞吐量，但使用场景有点不一样。

IO线程模型

BIO模型

单线程模型：只有一个线程处理，比如说聊天系统，同事只能有一个人聊天。现在谁还这么玩？
多线程模型：每个请求进来，系统都分配一个线程处理，请求数和线程数一比一，基本满足很小用户量使用。如果客户端请求过多，导致系统线程枯竭，整个系统可能就GG了。

socket = server.accept();
new HandlerThread(socket).run()

线程从socket请求到其他业务逻辑处理完都是单独一个线程处理。

对多线程模型的优化就是使用线程池，先创好建线程，一个用完后另一个接着用。因为线程的创建有系统开销。好比一个工具箱，这个人用完放回去其他人继续用。如果用完丢到炉子熔掉，其他人用再重新打造一个，开销就大了。使用线程池的另一个好处是，线程池有调度能力，可以用队列缓存请求，这样，吞吐量上来了，也不会因为无限制的创建线程导致挂掉。

 Socket socket = serverSocket.accept();
 executor.submit(new ConnectIOnHandler(socket));

bio线程模型图

BIO的缺点： 好比一个玩滑翔伞的景点，一天能接待多少玩家，完全取决于教练的数量。游客到了，必须有一个教练接待，带着一起飞。教练团就像一个线程池，飞完回来再接待另一个游客，但始终有限。

为什么会导致这样的问题。我们先看下一个BIO的socket实现的echoServer:

public class EchoServer {
    @Test
    public void socketServer() throws IOException {
        ExecutorService executor = Executors.newFixedThreadPool(100);
        ServerSocket serverSocket = new ServerSocket(9999);
        while (!Thread.currentThread().isInterrupted()) {
            Socket socket = serverSocket.accept();// 1.没有连接则阻塞等待
            executor.submit(new ConnectIOnHandler(socket));//为新的连接创建新的线程
        }
    }
    class ConnectIOnHandler extends Thread {
        private Socket socket;
        public ConnectIOnHandler(Socket socket) {
            this.socket = socket;
        }
        public void run() {
            while (!Thread.currentThread().isInterrupted() && !socket.isClosed()) {
                DataInputStream in = null;
                PrintStream out = null;
                try {
                     in = new DataInputStream(socket.getInputStream());
                     out = new PrintStream(socket.getOutputStream());
                    while (true) {
                        String line = in.readLine();// 2. 没有数据则阻塞等待
                        out.println(line); // 3. 写未就绪也阻塞
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }finally {
                    try {
                        if (in != null) {
                            in.close();
                        }
                        if (out != null) {
                            out.close();
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}

从示例中的1，2，3处注释可以看到，socket.accept()、socket.read()、socket.write()都是阻塞。网络IO是大爷，你问能不能读，他不告诉你，只能等有数据了才给你。这样，100万个请求进来，我得开100万个线程去等着。

一个快递员送快递，如果遇到一个客户不在家，就只能干等他回来，万一有个客户大晚上才回来，那快递员今天后面的其他快递就送不了，没办法，快递公司只能每一个快递配一个配送员。想想就不切实际，但BIO就是这样的。JAVA也不希望这样，但底层操作系统不支持其他方式，没办法啊。

如果能能打电话给客户，在不在家，给个回复，问题不就解决了。这就是引出了NIO。

NIO模型

上文说道，给个回复很是关键，NIO的关键就是：读写函数可以立刻返回。

NIO有个重要的抽象Selector，所有IO事件都往上面注册，它维护一个事件列表。 selector.select()会阻塞等去获取事件给感兴趣的chennel处理。这里的阻塞是没问题的，没有事件才阻塞，一般放入一个循环里执行。没有事件表明不需要分配任何线程去处理，当然得阻塞，否则CPU就空转啊。

像高速口售票员，没车时坐着，来车了收费起竿放过，一天一个人就能处理成千上万辆车。可能每辆车走完高速的时间没变，但吞吐量就上来了。这就是所谓性能的提升。

reactor模型

reactor模型得益于前面提到异步IO，因为读写不再阻塞，一个线程就可以处理处理所有网络IO事件，把具体逻辑分发给工作线程。

如图，reactor负责accept，read，write。读取数据后分配给工作线程处理，整个网络IO都是一个线程。

这个reactor的单线程模型，一个NIO线程负责监听和处理所有的客户端连接。不过网络通信不都这么容易，可能还有认证等逻辑，如果连接数太多，开销也不小。于是就有了主从reactor多线程模型：

mainReactor可以是一个线程池，只处理接收连接，后续的读写派发给subReactor，一定程度改善了性能。

而我们接下来要使用的netty框架其实就是Reactor的实现，了解Reactor，netty的线程模型解就比较清晰了。

具体可以参考Doug Lea的这个ppt:Scalable IO in Java。

netty实现rpc的线程设计

netty相对jdk的nio优点是，在后者基础上提供了更简洁的API，处理流程更清晰。

服务端

private EventLoopGroup bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("boss"));
private EventLoopGroup workGroup = new NioEventLoopGroup(
            Runtime.getRuntime().availableProcessors() * 2, new DefaultThreadFactory("worker-pool")
    );
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workGroup)
         .channel(NioServerSocketChannel.class)
         .childHandler(
                 new ChannelInitializer<NioSocketChannel>() {
                     @Override
                     protected void initChannel(NioSocketChannel ch) throws Exception {
                         ch.pipeline()
                           .addLast("decode", new Decoder())
                           .addLast("encode", new Encoder())
                           .addLast("severHandler", new NettyServerHandler());
                     }

                 }
         );
ChannelFuture future = bootstrap.bind(new InetSocketAddress(this.port));
future.syncUninterruptibly();
log.info(" started and listen on {}", future.channel().localAddress());

boss: 一个线程的线程池，从Reactor模型我们知道，mainReactor处理连接，一个线程就可以了。 work：读写转码，线程数一般是CPU的两倍。对rpc调用，可以把耗CPU的序列化移出，只处理IO读写。

客户端

rpc客户端可能调用上百个服务，不需要每次都启动一个连接，因为netty都是异步的，这些IO网络处理只需要一个workpool就行了，示例代码。

Bootstrap b = new Bootstrap();
b.group(clientWorkPool)
 .channel(NioSocketChannel.class)
 .handler(new ChannelInitializer<SocketChannel>() {
     @Override
     protected void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addLast("Decoder", new Decoder());
         ch.pipeline().addLast("Encoder", new Encoder());
         ch.pipeline().addLast("clientHandler", new NettyClientHandler());
     }
 });
ChannelFuture future = b.connect(address);
future.awaitUninterruptibly();
if (future.isSuccess() && future.channel().isActive()) {
    NettyClientHandler handler = future.channel().pipeline().get(NettyClientHandler.class);
    // 可以缓存下来，复用连接
    return handler;
}

参考文档： https://tech.meituan.com/nio.html http://www.infoq.com/cn/articles/netty-threading-model http://kasunpanorama.blogspot.hk/2015/04/understanding-reactor-pattern-with-java.html http://afghl.github.io/2016/12/17/java-nio-02-reactor-and-nio.html https://time.geekbang.org/column/article/4323