我发现了 高性能异步编程 和 实时流模型 那千丝万缕的联系！

本次我们来实现一个支持百万并发连接数的采集服务器，并且找到异步+ NIO 的编程方式和实时流模型的那些千丝万缕的联系。

一、需求说明

简单的需求功能如下：

• 数据采集是一个 web 服务器，可以接收 http 请求传上来的事件，事件是 json 格式的；
• 收到之后，进行解码，校验字符串是否可以解码成 json 对象；
• 对消息进行抽取、清洗、转化；
• 最后发送到 kafka 中。

性能要求：

• 支持百万并发连接数；
• 对 CPU 资源和 IO 资源充分利用；

二、实现方案剖析

网页和手机端会产生一些埋点文件，通过 http 方式发送给采集服务器。涉及到网络连接，第一个想到的就是 Socket。

1、初始版本：使用 BIO 实现的客户端和服务端通信

我们很容易就可以用多线程快速实现一个 web 端服务器，模型图如下（为了节省篇幅，代码我就不写了，很简单，但不实用，百度有很多）。

简单描述一下：每次来一个请求，都创建一个线程来执行。如下图：

但是弊端也显而易见，随随便便就能列出三点：

1.1 创建和销毁线程动作代价昂贵

Java 中的线程模型是基于操作系统原生线程模型实现的，也就是说 Java 中的线程其实是基于内核线程实现的，线程的创建，析构与同步都需要进行系统调用，而系统调用需要在用户态与内核中来回切换，代价相对较高；

1.2 线程本身占用大量的内存

Java 中，默认一个线程，线程栈大小是 1 M。一旦线程数过千，恐怕整个 Jvm 内存都会被吃掉一半；

1.3 线程切换成本是很高

操作系统在切换线程的时候，需要保留线程的上下文，然后再执行系统调用。如果线程数过多，可能执行线程的时间都会大于线程执行的时间，使系统陷于几乎不可用的状态；

2、改造版本：异步 + NIO 实现的高性能网络通信

这是 Java 中的 NIO 模型，如下图：

可能你一下无法接受同时出现这么多陌生的概念（Acceptor，Selector，Channel），没关系，NIO 再慢慢学，这里我们只要抓住它的核心：

• 使用了队列将请求接收器和工作线程隔开，让请求接收器和工作线程各自尽其所能的工作，更加充分的利用 IO 和 CPU 资源；
• 现在，NIO 连接器能够保持的并发连接数，不再受限于工作线程数量，无需分配大量线程，就能支持大量并发连接了。

3、进阶版本，如何充分利用 CPU 和 网络 IO 资源

在第二步，我们解决了高并发连接数的问题，但是还远远不够。

在一个采集系统中，我们需要做这三件事情，解码，清洗转化，发送。

其中，解码和清洗转化过程纯粹是 CPU 计算，占用 CPU 资源，而发送会大量占用 IO 资源。

如果让一个线程串行的执行这三件事，前面两件事，CPU 会很快做完，势必最后会等在 IO 操作上，这个线程就被操作系统挂起了，在那里空等，直到 IO 操作完成。

如何解决，只能增加工作线程数量，但是增加工作线程数量，会导致过多的线程调度和上下文切换，是另一种形式的 CPU 浪费。

如何解决，我们可以使用异步的方式。何谓异步，比如做饭过程就是异步，先把饭放电饭煲煮着，趁着这个时间去做菜，这就是“异步”。如果一直等到米饭煮熟，再去烧菜，这就是“同步”。

可能你就已经知道了，上次我们讲过 CompletableFuture，这是一个异步编程框架，可以将不同的线程编排起来。

并且可以指定线程池，让不同的事情，在不同的线程池里面完成。

看下面的代码：

// 解码线程池
Executor decoderExecutor = ExecutorHelper.createExecutor(2, "decoder");
// 转换线程池
Executor ectExecutor = ExecutorHelper.createExecutor(8, "ect");
// 发送线程池
Executor senderExecutor = ExecutorHelper.createExecutor(2, "sender");

@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpRequest req) throws Exception {
    CompletableFuture
   // 解码过程
            .supplyAsync(() -> this.decode(ctx, req), this.decoderExecutor)
   // 转换过程
            .thenApplyAsync(e -> this.doExtractCleanTransform(ctx, req, e), this.ectExecutor)
   // IO 过程
            .thenApplyAsync(e -> this.send(ctx, req, e), this.senderExecutor);
}

其中 channelRead0 是 Netty 框架的请求方法，每次请求都会到这个方法中。每次请求进来时，三个动作分别异步执行，但是 CompletableFuture 框架会保证每个请求三个动作执行的先后顺序。

这样，CPU 资源和 IO 资源，就可以得到充分的利用了。

三、进阶版本中存在的问题

1、问题描述

上述的异步模型可以用下面的图来表示

在上面的异步编程代码中，我们把不同类型的任务提交到不同的线程池中，而线程池是需要队列的，图上的队列就是线程池的队列。

其中，解码过程和转换过程，都是比较快速的过程，而发送的 I/O 过程则比较慢。

那么前面的消息会一直积压在发送过程的线程池队列中，等待执行。如果队列选择的是无界队列，那么越来越多的任务会积压，最终会用光所有的虚拟机所有的内存，导致 OOM。

2、如何控制事件的速度

我们可以直接想到，严格控制上游的发送速度，比如控制上游每秒钟只能发送 1000 条消息。这种方法虽然可行，但是非常不优雅。

如果下游遇到特殊原因，每秒只能处理 500 条，那仍然还是会 OOM，我们没法去估出一个合适的值的。

3、反向压力

有一种更加优雅的方案，叫做反向压力。

上面我们的方案出问题，主要原因还是在于队列是无界的，消息一直在积压，并且是非阻塞的。

要实现反向压力，只需要从两个方面来控制：

• 执行器的任务队列，它的容量必须是有限的；
• 当执行器的任务队列满了的时候，就阻止上游继续提交任务，直到任务队列中有新的空间为止。

如上图，可以看到，如果发送的线程池队列满了之后，就阻止上游的转换任务继续提交任务。过了一会，转换过程的队列也会满，同样的它也会阻止解码过程提交任务。

对于我们这种数据处理场景的话，可以通过横向增加服务器来解决 TPS 低的问题；如果是流式处理场景，那么最上游应该是主动从 Kafka 拉取消息，这个时候，它就放缓自己拉取消息的速度，从而做到流量控制。

当一段时间后，发送线程池队列有空闲了，又会继续处理消息。

4、实现反压

如何来实现反压？其实很简单，当队列满了之后，会进入线程池的拒绝策略中，在拒绝策略中，我们使用 while 循环来重复提交任务，直到任务提交成功，看下面的代码：

private final List<ExecutorService> executors;
private final Partitioner partitioner;
private Long rejectSleepMills = 1L;

public BackPressureExecutor(String name, int executorNumber, int coreSize, int maxSize, int capacity, long rejectSleepMills) {
    this.rejectSleepMills = rejectSleepMills;
    this.executors = new ArrayList<>(executorNumber);
    for (int i = 0; i < executorNumber; i++) {
        ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(capacity);
        this.executors.add(new ThreadPoolExecutor(
                coreSize, maxSize, 0L, TimeUnit.MILLISECONDS,
                queue,
                new ThreadFactoryBuilder().setNameFormat(name + "-" + i + "-%d").build(),
                new ThreadPoolExecutor.AbortPolicy()));
    }
    this.partitioner = new RoundRobinPartitionSelector(executorNumber);
}

@Override
public void execute(Runnable command) {
    boolean rejected;
    do {
        try {
           rejected = false;
           executors.get(partitioner.getPartition()).execute(command);
        } catch (RejectedExecutionException e) {
            rejected = true;
            try {
                TimeUnit.MILLISECONDS.sleep(rejectSleepMills);
            } catch (InterruptedException e1) {
                logger.warn("Reject sleep has been interrupted.", e1);
            }
        }
    } while (rejected);
}

可以看到上面的代码，关键点有两个：

第一个是，在创建 ThreadPoolExecutor 对象时，采用 ArrayBlockingQueue。这是一个容量有限的阻塞队列。因此，当任务队列已经满了时，就会停止继续往队列里添加新的任务，从而避免内存无限大，造成 OOM 问题。

第二个是，将 ThreadPoolExecutor 拒绝任务时，采用的策略设置为 AbortPolicy。这就意味着，在任务队列已经满了的时候，如果再向任务队列提交任务，就会抛出 RejectedExecutionException 异常。之后，我们再通过一个 while 循环，在循环体内，捕获 RejectedExecutionException 异常，并不断尝试，重新提交任务，直到成功为止。

这样，经过上面的改造，当下游的步骤执行较慢时，它的任务队列就会占满。这个时候，如果上游继续往下游提交任务，它就会不停重试。这样，自然而然地降低了上游步骤的处理速度，从而起到了流量控制的作用。

四、这不就是一个实时流模型！

上面的那个图，是不是似曾相识？没错，它就是实时流模型啊。

并且反压，已经成为流计算领域的共识，并且已经形成了反向压力相关的标准。

Flink 中是通过 Netty 的网络模型的阻塞来把压力一层层往上游传递的，和我们实现的这个有异曲同工之妙。

在如今的异步编程模型中，无处不存在着队列的影子，甚至在操作系统底层，也会使用队列来对性能做极致的优化，比如大名鼎鼎的 epoll。

而“队列”正是流计算系统最重要的组成结构。有“队列”的系统，它注定会是一个异步执行的过程，这也意味着“流”这种模式注定了是“异步”的。

五、总结

如今的分布式系统，都是几百甚至上千机器在一起协同工作，那不同的机器的不同进程一定会通信。

像 Spark，使用了 Netty 作为通信框架，Flink 也有在使用 Netty （还有Akka）作为通信框架。而要去了解一个分布式框架，第一步就是要了解它的通信框架，不然进程和进程的通信部分就没法看懂，整个框架核心逻辑也就无法透彻看懂。

而且在分布式系统中，异步编程的代码也是非常多，像 Flink 提交任务的过程，就使用了 CompletableFuture 异步编程框架来提交任务。