Java8函数式编程_9--使用Lambda表达式编写并发程序

1，免责声明，本文大部分内容摘自《Java8函数式编程》。在这本书的基础上，根据自己的理解和网上一些博文，精简或者修改。本次分享的内容，只用于技术分享，不作为任何商业用途。当然这本书是非常值得一读，强烈建议买一本！
2，本次分享的样例代码均上传到github上，请点击这里。

注意：本章所有的例子大多围绕 1.3 节介绍的案例展开(音乐)。

前面讨论了如何并行化处理数据，本章讨论如何使用 Lambda 表达式编写并发应用，高效传递信息和非阻塞式 I/O。

本章的一些例子用到了Vert.x 和 RxJava框架，但其中展现的设计原则是通用的，对其他框架或是自己编写的、没有使用任何框架的程序也适用。

主要内容如下：

9.1 为什么要使用非阻塞式I/O

9.2 回调

9.3 消息传递架构

9.4 末日金字塔

9.5 Future

9.6 CompletableFuture

9.7 响应式编程

9.8 何时何地使用新技术

9.9 要点回顾

9.1 为什么要使用非阻塞式I/O

在介绍并行化处理时，讲了很多关于如何高效利用多核 CPU 的内容。这种方式很管用，但在处理大量数据时，它并不是唯一可用的线程模型。

假设要编写一个支持大量用户的聊天程序。每当用户连接到聊天服务器时，都要和服务器建立一个 TCP 连接。使用传统的线程模型，每次向用户写数据时，都要调用一个方法向用户传输数据，这个方法会阻塞当前线程。

这种 I/O 方式叫阻塞式 I/O，是一种通用且易于理解的方式，因为和程序用户的交互通常符合这样一种顺序执行的方式。缺点是，将系统扩展至支持大量用户时，需要和服务器建大量 TCP 连接，因此扩展性不是很好。

非阻塞式 I/O，有时也叫异步I/O，可以处理大量并发网络连接，而且一个线程可以为多个连接服务。和阻塞式 I/O 不同，对聊天程序客户端的读写调用立即返回，真正的读写操作则在另一个独立的线程执行，这样就可以同时执行其他任务了。如何使用这些省下来的 CPU 周期完全取决于程序员，可以选择读入更多数据，也可以玩一局游戏。

到目前为止，这里避免使用代码来描述这两种 I/O 方式，因为根据 API 的不同，它们有多种实现方式。Java 标准类库的 NIO 提供了非阻塞式 I/O 的接口，NIO 的最初版本用到了 Selector 的概念，让一个线程管理多个通信管道，比如：向客户端写数据的网络套接字。

然而这种方式压根儿就没有在 Java 程序员中流行起来，它编写出来的代码难于理解和调试。引入 Lambda 表达式后，设计和实现没有这些缺点的 API 就顺手多了。

9.2 回调

为了展示非阻塞式 I/O 的原则，我们将运行一个极其简单的聊天应用，没有那些花里胡哨的功能。当用户第一次连接应用时，需要设定用户名，随后便可通过应用收发信息。

我们将使用 Vert.x 框架实现该应用，并且在实施过程中根据需要，引入其他一些必需的技术。让我们先来写一段接收 TCP 连接的代码，如例 9-1 所示。

// 例 9-1 接收 TCP 连接

public class ChatVerticle extends Verticle {public void start() { vertx.createNetServer().connectHandler(socket -> {container.logger().info("socket connected");socket.dataHandler(new User(socket, this));}).listen(10_000);container.logger().info("ChatVerticle started");}
}

读者可Verticle 类想成 Servlet—— 它是 Vert.x 框架中部署的原子单元。上述代码的入口是 start方法，它和普通 Java 程序中的 main 方法类似。在聊天应用中，我们用它建立一个接收 TCP 连接的服务器。

然后向 connectHandler 方法输入一个 Lambda 表达式，每当有用户连接到聊天应用时，都会调用该 Lambda 表达式。这就是一个回调，与在第 1 章中介绍的 Swing 中的回调类似。这种方式的好处是，应用不必控制线程模型——Vert.x 框架为我们管理线程，打理好了一切相关复杂性，程序员只需考虑事件和回调就够了。

我们的应用还通过 dataHandler 方法注册了另外一个回调，每当从网络套接字读取数据时，该回调就会被调用。在本例中，我们希望提供更复杂的功能，因此没有使用 Lambda 表达式，而是传入一个常规的 User 类，该类实现了相关的函数接口。User 类的定义如例 9-2 所示。

// 例 9-2 处理用户连接

public class User implements Handler<Buffer> {private static final Pattern newline = Pattern.compile("\\n");private final NetSocket socket;private final EventBus eventBus;private Optional<String> name;public User(NetSocket socket, Verticle verticle) {Vertx vertx = verticle.getVertx();this.socket = socket;eventBus = vertx.eventBus();name = Optional.empty();}@Overridepublic void handle(Buffer buffer) {newline.splitAsStream(buffer.toString()).forEach(line -> {if (!name.isPresent()) {setName(line);} else {handleMessage(line);}});}public Optional<String> getName() {return name;}public void setName(String name) {this.name = Optional.of(name);}private void handleMessage(String message) {System.out.println("handle message: " + message);}}// Class continues...

变量 buffer 包含了网络连接写入的数据，我们使用的是一个分行的文本协议，因此需要先将其转换成一个字符串，然后依换行符分割。

Java 8为 Pattern 类新增了一个 splitAsStream 方法，该方法使用正则表达式将字符串分割好后，生成一个包含分割结果的流对象。

用户连上聊天服务器后，首先要做的事是设置用户名。如果用户名未知，则执行设置用户名的逻辑；否则正常处理聊天消息。

还需要接收来自其他用户的消息，并且将它们传递给聊天程序客户端，让接收者能够读取消息。为了实现该功能，在设置当前用户用户名的同时，我们注册了另外一个回调，用来写入消息(例 9-3)。

// 例 9-3 注册聊天消息

eventBus.registerHandler(name, (Message<String> msg) -> {sendClient(msg.body());
});

上述代码使用了 Vert.x 的事件总线，它允许在 verticle 对象之间以非阻塞式 I/O 的方式传递消息(如图 9-1 所示)。registerHandler 方法将一个处理程序和一个地址关联，有消息发送给该地址时，就将之作为参数传递给处理程序，并且自动调用处理程序。这里使用用户名作为地址。

// 图 9-1:使用事件总线传递消息

通过为地址注册处理程序并发消息的方式，可以构建非常复杂和解耦的服务，它们之间完全以非阻塞式 I/O 方式响应。

Vert.x 的事件总线允许发送多种类型的消息，但是它们都要使用 Message 对象进行封装。点对点的消息传递由 Message 对象本身完成，它们可能持有消息发送方的应答处理程序。在这种情况下，我们想要的是消息体，也就是文字本身，则只需调用 body 方法。我们通过将消息写入 TCP 连接，把消息发送给了用户聊天客户端。

当应用想要把消息从一个用户发送给另一个用户时，就使用代表另一个用户的地址(如例 9-4 所示)，这里使用了用户的用户名。

// 例 9-4 发送聊天信息

eventBus.send(user, name.get() +‘>’+ message);

让我们扩展这个基础聊天服务器，向关注你的用户群发消息，为此，需要实现两个新命令。代表群发命令的感叹号，它能将信息群发给关注你的用户。如果 Bob 键入 “!hello followers”，则所有关注 Bob 的用户都会收到该条信息：“Bob>hello followers”。