前言

Goroutines 是 Go 语言主要的并发原语。它看起来非常像线程，但是相比于线程它的创建和管理成本很低。Go 在运行时将 goroutine 有效地调度到真实的线程上，以避免浪费资源，因此您可以轻松地创建大量的 goroutine（例如每个请求一个 goroutine），并且您可以编写简单的，命令式的阻塞代码。因此，Go 的网络代码往往比其它语言中的等效代码更直接，更容易理解（这点从下文中的示例代码可以看出）。

对我来说，goroutine 是将 Go 这门语言与其它语言区分开来的一个主要特征。这就是为什么大家更喜欢用 Go 来编写需要并发的代码。在下面讨论更多关于 goroutine 之前，我们先了解一些历史，这样你就能理解为什么你想要它们了。

基于 fork 和线程

高性能服务器需要同时处理来自多个客户端的请求。有很多方法可以设计一个服务端架构来处理这个问题。最容易想到的就是让一个主进程在循环中调用 accept，然后调用 fork 来创建一个处理请求的子进程。这篇 Beej’s Guide to Network Programming 指南中提到了这种方式。

在网络编程中，fork 是一个很好的模式，因为你可以专注于网络而不是服务器架构。但是它很难按照这种模式编写出一个高效的服务器，现在应该没有人在实践中使用这种方式了。

fork 同时也存在很多问题，首先第一个是成本: Linux 上的 fork 调用看起来很快，但它会将你所有的内存标记为 copy-on-write。每次写入 copy-on-write 页面都会导致一个小的页面错误，这是一个很难测量的小延迟，进程之间的上下文切换也很昂贵。

另一个问题是规模: 很难在大量子进程中协调共享资源（如 CPU、内存、数据库连接等）的使用。如果流量激增，并且创建了太多进程，那么它们将相互争夺 CPU。但是如果限制创建的进程数量，那么在 CPU 空闲时，大量缓慢的客户端可能会阻塞每个人的正常使用，这时使用超时机制会有所帮助（无论服务器架构如何，超时设置都是很必要的）。

通过使用线程而不是进程，上面这些问题在一定程度上能得到缓解。创建线程比创建进程更“便宜”，因为它共享内存和大多数其它资源。在共享地址空间中，线程之间的通信也相对容易，使用信号量和其它结构来管理共享资源，然而，线程仍然有很大的成本，如果你为每个连接创建一个新线程，你会遇到扩展问题。与进程一样，你此时需要限制正在运行的线程的数量，以避免严重的 CPU 争用，并且需要使慢速请求超时。创建一个新线程仍然需要时间，尽管可以通过使用线程池在请求之间回收线程来缓解这一问题。

无论你是使用进程还是线程，你仍然有一个难以回答的问题: 你应该创建多少个线程？如果您允许无限数量的线程，客户端可能会用完所有的内存和 CPU，而流量会出现小幅激增。如果你限制服务器的最大线程数，那么一堆缓慢的客户端就会阻塞你的服务器。虽然超时是有帮助的，但它仍然很难有效地使用你的硬件资源。

基于事件驱动

那么既然无法轻易预测出需要多少线程，当如果尝试将请求与线程解耦时会发生什么呢？如果我们只有一个线程专门用于应用程序逻辑（或者可能是一个小的、固定数量的线程），然后在后台使用异步系统调用处理所有的网络流量，会怎么样？这就是一种 事件驱动 的服务端架构。

事件驱动架构模式是围绕 select 系统调用设计的。后来像 poll 这样的机制已经取代了 select，但是 select 是广为人知的，它们在这里都服务于相同的概念和目的。select 接受一个文件描述符列表（通常是套接字），并返回哪些是准备好读写的。如果所有文件描述符都没有准备好，则选择阻塞，直到至少有一个准备好。

1
2
3
4
5
6
7
8
9
10
11
12

#include <sys/select.h>
#include <poll.h>

int select(int nfds, 
           fd_set *restrict readfds, 
           fd_set *restrict writefds, 
           fd_set *restrict exceptfds, 
           struct timeval *restrict timeout);

int poll(struct pollfd *fds, 
         nfds_t nfds, 
         int timeout);

为了实现一个事件驱动的服务器，你需要跟踪一个 socket 和网络上被阻塞的每个请求的一些状态。在服务器上有一个单一的主事件循环，它调用 select 来处理所有被阻塞的套接字。当 select 返回时，服务器知道哪些请求可以进行了，因此对于每个请求，它调用应用程序逻辑中的存储状态。当应用程序需要再次使用网络时，它会将套接字连同新状态一起添加回“阻塞”池中。这里的状态可以是应用程序恢复它正在做的事情所需的任何东西: 一个要回调的 closure，或者一个 Promise。

从技术上讲，这些其实都可以用一个线程实现。这里不能谈论任何特定实现的细节，但是像 JavaScript
这样缺乏线程的语言也很好的遵循了这个模型。Node.js 更是将自己描述为“an event-driven JavaScript runtime, designed to build scalable network applications.”

事件驱动的服务器通常比纯粹基于 fork 或线程的服务器更好地利用 CPU 和内存。你可以为每个核心生成一个应用程序线程来并行处理请求。线程不会相互争夺 CPU，因为线程的数量等于内核的数量。当有请求可以进行时，线程永远不会空闲，非常高效。效率如此之高，以至于现在大家都使用这种方式来编写服务端代码。

从理论上讲，这听起来不错，但是如果你编写这样的应用程序代码，就会发现这是一场噩梦。。。具体是什么样的噩梦，取决于你所使用的语言和框架。在 JavaScript 中，异步函数通常返回一个 Promise，你给它附加回调。在 Java gRPC 中，你要处理的是 StreamObserver。如果你不小心，你最终会得到很多深度嵌套的“箭头代码”函数。如果你很小心，你就把函数和类分开了，混淆了你的控制流。不管怎样，你都是在 callback hell 里。

下面是一个 Java gRPC 官方教程 中的一个示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

public void routeChat() throws Exception {
  info("*** RoutChat");
  final CountDownLatch finishLatch = new CountDownLatch(1);
  StreamObserver<RouteNote> requestObserver =
      asyncStub.routeChat(new StreamObserver<RouteNote>() {
        @Override
        public void onNext(RouteNote note) {
          info("Got message \"{0}\" at {1}, {2}", note.getMessage(), note.getLocation()
              .getLatitude(), note.getLocation().getLongitude());
        }

        @Override
        public void onError(Throwable t) {
          Status status = Status.fromThrowable(t);
          logger.log(Level.WARNING, "RouteChat Failed: {0}", status);
          finishLatch.countDown();
        }

        @Override
        public void onCompleted() {
          info("Finished RouteChat");
          finishLatch.countDown();
        }
      });

  try {
    RouteNote[] requests =
        {newNote("First message", 0, 0), newNote("Second message", 0, 1),
            newNote("Third message", 1, 0), newNote("Fourth message", 1, 1)};

    for (RouteNote request : requests) {
      info("Sending message \"{0}\" at {1}, {2}", request.getMessage(), request.getLocation()
          .getLatitude(), request.getLocation().getLongitude());
      requestObserver.onNext(request);
    }
  } catch (RuntimeException e) {
    // Cancel RPC
    requestObserver.onError(e);
    throw e;
  }
  // Mark the end of requests
  requestObserver.onCompleted();

  // Receiving happens asynchronously
  finishLatch.await(1, TimeUnit.MINUTES);
}

上面代码官方的初学者教程，它不是一个完整的例子，发送代码是同步的，而接收代码是异步的。在 Java 中，你可能会为你的 HTTP 服务器、gRPC、数据库和其它任何东西处理不同的异步类型，你需要在所有这些服务器之间使用适配器，这很快就会变得一团糟。

同时这里如果使用锁也很危险，你需要小心跨网络调用持有锁。锁和回调也很容易犯错误。例如，如果一个同步方法调用一个返回 ListenableFuture 的函数，然后附加一个内联回调，那么这个回调也需要一个同步块，即使它嵌套在父方法内部。

Goroutines

终于到了我们的主角——goroutines。它是 Go 语言版本的线程。像它语言（比如：Java）中的线程一样，每个 gooutine 都有自己的堆栈。goroutine 可以与其它 goroutine 并行执行。与线程不同，goroutine 的创建成本非常低:它不绑定到 OS 线程上，它的堆栈开始非常小（初始只有 2 K），但可以根据需要增长。当你创建一个 goroutine 时，你实际上是在分配一个 closure，并在运行时将其添加到队列中。

在内部实现中，Go 的运行时有一组执行程序的 OS 线程（通常每个内核一个线程）。当一个线程可用并且一个 goroutine 准备运行时，运行时将这个 goroutine 调度到线程上，执行应用程序逻辑。如果一个运行例程阻塞了像 mutex 或 channel 这样的东西时，运行时将它添加到阻塞的运行 goroutine 集合中，然后将下一个就绪的运行例程调度到同一个 OS 线程上。

这也适用于网络:当一个线程程序在未准备好的套接字上发送或接收数据时，它将其 OS 线程交给调度器。这听起来是不是很熟悉？Go 的调度器很像事件驱动服务器中的主循环。除了仅仅依赖于 select 和专注于文件描述符之外，调度器处理语言中可能阻塞的所有内容。

你不再需要避免阻塞调用，因为调度程序可以有效地利用 CPU。可以自由地生成许多 goroutine（可以每个请求一个!），因为创建它们的成本很低，而且不会争夺 CPU，你不需要担心线程池和执行器服务，因为运行时实际上有一个大的线程池。

简而言之，你可以用干净的命令式风格编写简单的阻塞应用程序代码，就像在编写一个基于线程的服务器一样，但你保留了事件驱动服务器的所有效率优势，两全其美。这类代码可以很好地跨框架组合。你不需要 streamobserver 和 ListenableFutures 之间的这类适配器。

下面让我们看一下来自 Go gRPC 官方教程 的相同示例。可以发现这里的控制流比 Java 示例中的更容易理
解，因为发送和接收代码都是同步的。在这两个 goroutines 中，我们都可以在一个 for 循环中调用 stream.Recv 和stream.Send。不再需要回调、子类或执行器这些东西了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

stream, err := client.RouteChat(context.Background())
waitc := make(chan struct{})
go func() {
  for {
    in, err := stream.Recv()
    if err == io.EOF {
      // read done.
      close(waitc)
      return
    }
    if err != nil {
      log.Fatalf("Failed to receive a note : %v", err)
    }
    log.Printf("Got message %s at point(%d, %d)", in.Message, in.Location.Latitude, in.Location.Longitude)
  }
}()
for _, note := range notes {
  if err := stream.Send(note); err != nil {
    log.Fatalf("Failed to send a note: %v", err)
  }
}
stream.CloseSend()
<-waitc

虚拟线程

如何你使用 Java 这门语言，到目前为止，你要么必须生成数量不合理的线程，要么必须处理 Java 特有的回调地狱。令人高兴的是，JEP 444 中增加了 virtual threads，这看起来很像 Go 语言中的 goroutine。

创建虚拟线程的成本很低。JVM 将它们调度到平台线程（platform threads，内核中的真实线程）上。平台线程的数量是固定的，一般每个内核一个平台线程。当一个虚拟线程执行阻塞操作时，它会释放它的平台线程，JVM
可能会将另一个虚拟线程调度到它上面。与 gooutine 不同，虚拟线程调度是协作的: 虚拟线程在执行阻塞操作之前不会服从于调度程序。这意味着紧循环可以无限期地保持线程。目前不清楚这是实现限制还是有更深层次的问题。Go 以前也有这个问题，直到 1.14 才实现了完全抢占式调度（可见 GopherCon 2021）。

Java 的虚拟线程现在可以预览，预计在 JDK 21 中成为 stable（官方消息是预计 2023 年 9 月发布）状态。哈哈，很期待到时候能删除大量的 ListenableFutures。每当引入一种新的语言或运行时特性时，都会有一个漫长的迁移过渡期，个人认为 Java 生态系统在这方面还是过于保守了。

前言

首先需要明确的是 TCP 是一个可靠传输协议，它的所有特点最终都是为了这个可靠传输服务。在网上看到过很多文章讲 TCP 连接的三次握手和断开连接的四次挥手，但是都太过于理论，看完感觉总是似懂非懂。反复思考过后，觉得我自己还是偏工程型的人，要学习这些理论性的知识，最好的方式还是要通过实际案例来理解，这样才会具象深刻。本文通过 Wireshark 抓包来分析 TCP 三次握手和四次挥手，如果你也对这些理论感觉似懂非懂，那么强烈建议你也结合抓包实践来强化理解这些理论性的知识。

三次握手

TCP 建立连接的三次握手是连接的双方协商确认一些信息（Sequence number、Maximum Segment Size、Window Size 等），Sequence number 有两个作用：一个是 SYN 标识位为 1 时作为初始序列号（ISN），则实际第一个数据字节的序列号和相应 ACK 中的确认号就是这个序列号加 1；另一个是 SYN 标识位为 0 时，则是当前会话的 segment（传输层叫 segment，网络层叫 packet，数据链路层叫 frame）的第一个数据字节的累积序列号。Maximum Segment Size 简称 MSS，表示最大一个 segment 中能传输的信息（不含 TCP、IP 头部）。Window Size 表示发送方接收窗口的大小。下面看看我在本地访问博客 mghio 的三次握手过程：

图中三个小红框表示与服务器建立连接的三次握手。

1. 第一步，client 端（这个示例也就是浏览器）发送 SYN 到 server 端；
2. 第二步，server 端收到 SYN 消息后，回复 SYN + ACK 到client 端，ACK 表示已经收到了 client 的 SYN 消息；
3. 第三步，client 端收到回复 SYN + ACK 后，也回复一个 ACK 表示收到了 server 端的 SYN + ACK 了，其实

到这一步，client 端的 60469 端口已经是 ESTABLISHED 状态了。
可以看到，其实三次握手的核心目的就是双方互相告知对象自己的 Sequence number，蓝框是 client 端的初始 Sequence number 和 client 端回复的 ACK，绿框是 server 端的初始 Sequence number 和 client 端回复的 ACK。这样协商好初始 Sequence number 后，发送数据包时发送端就可以判断丢包和进行丢包重传了。

三次握手还有一个目的是协商一些信息（上图中黄色方框是 Maximum Segment Size，粉色方框是 Window Size）。

到这里，就可以知道平常所说的建立TCP连接本质是为了实现 TCP 可靠传输做的前置准备工作，实际上物理层并没有这个连接在那里。TCP 建立连接之后时拥有和维护一些状态信息，这个状态信息就包含了 Sequence number、MSS、Window Size 等，TCP 握手就是协商出来这些初始值。而这些状态才是我们平时所说的 TCP 连接的本质。因为这个太重要了，我还要再次强调一下，TCP 是一个可靠传输协议，它的所有特点最终都是为了这个可靠传输服务。

四次挥手

下面再来看看，当关闭浏览器页面是发生断开连接的四次挥手过程:

相信你已经发现了，上图抓包抓到的不是四次挥手，而是三次挥手，这是为何呢？

这是由于 TCP 的时延机制（因为系统内核并不知道应用能不能立即关闭），当被挥手端（这里是 server 的 443 端口）第一次收到挥手端（这里是 client 的 63612 端口）的 FIN 请求时，并不会立即发送 ACK，而是会经过一段延迟时间后再发送，但是此时被挥手端也没有数据发送，就会向挥手端发送 FIN 请求，这里就可能造成被挥手端发送的 FIN 与 ACK 一起被挥手端收到，导致出现第二、三次挥手合并为一次的现象，也就最终呈现出“三次挥手”的情况。

断开连接四次挥手分为如下四步（假设没有出现挥手合并的情况）：

1. 第一步，client 端主动发送 FIN 包给 server 端；
2. 第二步，server 端回复 ACK（对应第一步 FIN 包的 ACK）给 client，表示 server 知道 client 端要断开了；
3. 第三步，server 端发送 FIN 包给 client 端，表示 server 端也没有数据要发送了，可以断开了；
4. 第四步，client 端回复 ACK 包给 server 端，表示既然双发都已发送 FIN 包表示可以断开，那么就真的断开了啊。

下面是 TCP 连接流转状态图（其中 CLOSED 状态是虚拟的，实际上并不存在），这个图很重要，记住这个图后基本上所有的 TCP 网络问题就可以解决。

其中比较难以理解的是 TIME_WAIT 状态，主动关闭的那一端会经历这个状态。这一端停留在这个状态的最长时间是 Maximum segment lifetime（MSL）的 2 倍，大部分时候被简称之为 2MSL。存在 TIME_WAIT 状态有如下两个原因：

1. 要可靠的实现 TCP 全双工连接终止；
2. 让老的重复 segment 在网络中消失（一个 sement 在网络中存活的最长时间为 1 个 MSL，一来一回就是 2 MSL）；

为什么握手是三次，而挥手是四次？

嘿嘿，这是个经典的面试题，其实大部分人都背过挥手是四次的原因：因为 TCP 是全双工（双向）的，所以回收需要四次……。但是再反问下：握手也是双向的，但是为什么是只要三次呢？

网上流传的资料都说 TCP 是双向的，所以回收需要四次，但是握手也是双向（握手双方都在告知对方自己的初始 Sequence number），那么为什么就不用四次握手呢？所以凡事需要多问几个为什么，要有探索和怀疑精神。

你再仔细回看上面三次握手的第二步（SYN + ACK），其实是可以拆分为两步的：第一步回复 ACK，第二步再发 SYN 也是完全可以的，只是效率会比较低，这样的话三次握手不也变成四次握手了。

看起来四次挥手主要是收到第一个 FIN 包后单独回复了一个 ACK 包这里多了一次，如果能像握手那样也回复 FIN + ACK 那么四次挥手也就变成三次了。这里再贴一下上面这个挥手的抓包图：

这个图中第二个红框就是 server 端回复的 FIN + ACK 包，这样四次挥手变成三次了（如果一个包算一次的话）。这里使用四次挥手原因主要是：被动关闭端在收到 FIN 后，知道主动关闭端要关闭了，然后系统内核层会通知应用层要关闭，此时应用层可能还需要做些关闭前的准备工作，可能还有数据没发送完，所以系统内核先回复一个 ACK 包，然后等应用层准备好了主动调 close 关闭时再发 FIN 包。

而握手过程中就没有这个准备过程了，所以可以立即发送 SYN + ACK（在这里的两步合成一步了，提高效率）。挥手过程中系统内核在收到对方的 FIN 后，只能 ACK，不能主动替应用来 FIN，因为系统内核并不知道应用能不能立即关闭。

总结

TCP 是一个很复杂的协议，为了实现可靠传输以及处理各种网络传输中的 N 多问题，有一些很经典的解决方案，比如其中的网络拥塞控制算法、滑动窗口、数据重传等。强烈建议你去读一下 rfc793 和 TCP/IP 详解 卷1：协议 这本书。

如果你是那些纯看理论就能掌握好一门技能，然后还能举三反一的人，那我很佩服你；如果不是，那么学习理论知识注意要结合实践来强化理解理论，要经过反反复复才能比较好地掌握一个知识，讲究技巧，必要时要学会通过工具来达到目的。

最后 TCP 所有特性基本上核心都是为了实现可靠传输这个目标来服务的，然后有一些是出于优化性能的目的。

前言

在 Spring 框架中有很多实用的功能，不需要写大量的配置代码，只需添加几个注解即可开启。 其中一个重要原因是那些 @EnableXXX 注解，它可以让你通过在配置类加上简单的注解来快速地开启诸如事务管理（@EnableTransactionManagement）、Spring MVC（@EnableWebMvc）或定时任务（@EnableScheduling）等功能。这些看起来简单的注解语句提供了很多功能，但它们的内部机制从表面上看却不太明显。 一方面，对于使用者来说用这么少的代码获得这么多实用的功能是很好的，但另一方面，如果你不了解某个东西的内部是如何工作的，就会使调试和解决问题更加困难。

设计目标

Spring 框架中那些 @EnableXXX 注解的设计目标是允许用户用最少的代码来开启复杂使用的功能。 此外，用户必须能够使用简单的默认值，或者允许手动配置该代码。最后，代码的复杂性要向框架使用者隐藏掉。 简而言之，让使用者设置大量的 Bean，并选择性地配置它们，而不必知道这些 Bean 的细节（或真正被设置的内容）。下面来看看具体的几个例子：

@EnableScheduling (导入一个 @Configuration 类)

首先要知道的是，@EnableXXX 注解并不神奇。实际上在 BeanFactory 中并不知道这些注解的具体内容，而且在 BeanFactory 类中，核心功能和特定注解（如 @EnableWebMvc）或它们所存放的 jar 包（如 spring-web）之间没有任何依赖关系。 让我们看一下 @EnableScheduling，下面看看它是如何工作的。 定义一个 SchedulingConfig 配置类，如下所示：

1
2
3
4
5

@Configuration
@EnableScheduling
public class SchedulingConfig {
  // some bean in here
}

上面的内容没有什么特别之处。只是一个用 @EnableScheduling 注释的标准 Java 配置。@EnableScheduling 让你以设定的频率执行某些方法。例如，你可以每 10 分钟运行 BankService.transferMoneyToMghio()。 @EnableScheduling 注解源码如下：

1
2
3
4
5
6
7

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Import(SchedulingConfiguration.class)
@Documented
public @interface EnableScheduling {

}

上面的 EnableScheduling 注解，我们可以看到它只是一个标准的类级注解（@Target/@Retention），应该包含在 JavaDocs 中（@Documented），但是它有一个 Spring 特有的注解（@Import）。 @Import 是将一切联系起来的关键。 在这种情况下，由于我们的 SchedulingConfig 被注解为 @EnableScheduling，当 BeanFactory 解析文件时（内部是ConfigurationClassPostProcessor 在解析它），它也会发现 @Import(SchedulingConfiguration.class) 注解，它将导入该值中定义的类。 在这个注解中，就是 SchedulingConfiguration。

这里导入是什么意思呢？在这种情况下，它只是被当作另一个 Spring Bean。 SchedulingConfiguration 实际上被注解为@Configuration，所以 BeanFactory 会把它看作是另一个配置类，所有在该类中定义的 Bean 都会被拉入你的应用上下文，就像你自己定义了另一个 @Configuration 类一样。 如果我们检查 SchedulingConfiguration，我们可以看到它只定义了一个Bean（一个Post Processor），它负责我们上面描述的调度工作，源码如下：

1
2
3
4
5
6
7
8
9
10
11

@Configuration
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public class SchedulingConfiguration {

@Bean(name = TaskManagementConfigUtils.SCHEDULED_ANNOTATION_PROCESSOR_BEAN_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public ScheduledAnnotationBeanPostProcessor scheduledAnnotationProcessor() {
return new ScheduledAnnotationBeanPostProcessor();
}

}

也许你会问，如果想配置 SchedulingConfiguration 中定义的 bean 呢？ 这里也只是在处理普通的Bean。 所以你对其它 Bean 所使用的机制也适用于此。 在这种情况下，ScheduledAnnotationBeanPostProcessor 使用一个标准的 Spring Bean 生命周期（postProcessAfterInitialization）来发现应用程序上下文何时被刷新。 当符合条件时，它会检查是否有任何 Bean 实现了 SchedulingConfigurer，如果有，就使用这些 Bean 来配置自己。 其实这一点并不明细（在 IDE 中也不太容易找到），但它与 BeanFactory 是完全分离的，而且是一个相当常见的模式，一个 Bean 被用来配置另一个 Bean。 而现在我们可以把所有的点连接起来，它（在某种程度上）很容易找到（你可以 Google 一下文档或阅读一下 JavaDocs）。

@EnableTransactionManagement（导入一个 ImportSelector）

在上一个示例中，我们讨论了像 @EnableScheduling 这样的注解如何使用 @Import 来导入另一个 @Configuration 类并使其所有的 Bean 对你的应用程序可用（和可配置）。但是如果你想根据某些配置加载不同的 Bean 集，会发生什么呢？ @EnableTransactionManagement 就是一个很好的例子。TransactioConfig 定义如下：

1
2
3
4
5

@Configuration
@EnableTransactionManagement(mode = AdviceMode.ASPECTJ)
public class TransactioConfig {
    // some bean in here
}

再一次，上面没有什么特别之处。只是一个用@EnableTransactionManagement注释的标准Java配置。唯一与之前的例子有些不同的是，用户为注释指定了一个参数（mode=AdviceMode.ASPECTJ）。 @EnableTransactionManagement注解本身看起来像这样。

1
2
3
4
5
6
7
8
9
10
11
12

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(TransactionManagementConfigurationSelector.class)
public @interface EnableTransactionManagement {

boolean proxyTargetClass() default false;

AdviceMode mode() default AdviceMode.PROXY;

int order() default Ordered.LOWEST_PRECEDENCE;
}

和前面一样，一个相当标准的注解，尽管这次它有一些参数。 然而，正如前文提到，@Import 注解是将一切联系在一起的关键，这一点再次得到证实。 但区别在于，这次我们导入的是 TransactionManagementConfigurationSelector 这个类，通过源码可以发现，其实它不是一个被 @Configuration 注解的类。 TransactionManagementConfigurationSelector 是一个实现ImportSelector 的类。 ImportSelector 的目的是让你的代码选择在运行时加载哪些配置类。 它有一个方法，接收关于注解的一些元数据，并返回一个类名数组。 在这种情况下，TransactionManagementConfigurationSelector 会查看模式并根据模式返回一些类。其中的 selectImports 方法源码如下：

1
2
3
4
5
6
7
8
9
10
11
12

@Override
protected String[] selectImports(AdviceMode adviceMode) {
  switch (adviceMode) {
    case PROXY:
      return new String[] {AutoProxyRegistrar.class.getName(),
                           ProxyTransactionManagementConfiguration.class.getName()};
    case ASPECTJ:
      return new String[] {determineTransactionAspectClass()};
    default:
      return null;
  }
}

这些类中的大多数是 @Configuration（例如 ProxyTransactionManagementConfiguration），通过前文介绍我们知道它们会像前面一样工作。 对于 @Configuration 类，它们被加载和配置的方式与我们之前看到的完全一样。 所以简而言之，我们可以使用 @Import 和 @Configuration 类来加载一套标准的 Bean，或者使用 @Import 和 ImportSelector 来加载一套在运行时决定的 Bean。

@EnableAspectJAutoProxy (在 Bean 定义层导入)

@Import 支持的最后一种情况，即当你想直接处理 BeanRegistry（工厂）时。如果你需要操作Bean Factory或者在Bean定义层处理Bean，那么这种情况就适合你，它与上面的情况非常相似。 你的 AspectJProxyConfig 可能看起来像。

1
2
3
4
5

@Configuration
@EnableAspectJAutoProxy 
public class AspectJProxyConfig {
  // some bean in here
}

再一次，上面定义没有什么特别的东西。只是一个用 @EnableAspectJAutoProxy 注释的标准 Java 配置。 下面是@EnableAspectJAutoProxy 的源代码。

1
2
3
4
5
6
7
8
9
10

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(AspectJAutoProxyRegistrar.class)
public @interface EnableAspectJAutoProxy {

boolean proxyTargetClass() default false;

boolean exposeProxy() default false;
}

和前面一样，@Import 是关键，但这次它指向 AspectJAutoProxyRegistrar，它既没有 @Configuration 注解，也没有实现 ImportSelector 接口。 这次使用的是实现了 ImportBeanDefinitionRegistrar。 这个接口提供了对 Bean 注册中心（Bean Registry）和注解元数据的访问，因此我们可以在运行时根据注解中的参数来操作 Bean 注册表。 如果你仔细看过前面的示例，你可以看到我们忽略的类也是 ImportBeanDefinitionRegistrar。 在 @Configuration 类不够用的时候，这些类会直接操作 BeanFactory。

所以现在我们已经涵盖了 @EnableXXX 注解使用 @Import 将各种 Bean 引入你的应用上下文的所有不同方式。 它们要么直接引入一组 @Configuration 类，这些类中的所有 Bean 都被导入到你的应用上下文中。 或者它们引入一个 ImportSelector 接口实现类，在运行时选择一组 @Configuration 类并将这些 Bean 导入到你的应用上下文中。 最后，他们引入一个ImportBeanDefinitionRegistrars，可以直接与 BeanFactory 在 BeanDefinition 级别上合作。

结论

总的来说，个人认为这种将 Bean 导入应用上下文的方法很好，因为它使框架使用者的使用某个功能非常容易。不幸的是，它模糊了如何找到可用的选项以及如何配置它们。 此外，它没有直接利用 IDE 的优势，所以很难知道哪些 Bean 正在被创建（以及为什么）。 然而，现在我们知道了 @Import 注解，我们可以使用 IDE 来挖掘一下每个注解及其相关的配置类，并了解哪些 Bean 正在被创建，它们如何被添加到你的应用上下文中，以及如何配置它们。 希望对你有帮助~

前言

在并发编程中，当多个线程同时访问同一个共享的可变变量时，会产生不确定的结果，所以要编写线程安全的代码，其本质上是对这些可变的共享变量的访问操作进行管理。导致这种不确定结果的原因就是可见性、有序性和原子性问题，Java 为解决可见性和有序性问题引入了 Java 内存模型，使用互斥方案（其核心实现技术是锁）来解决原子性问题。这篇先来看看解决可见性、有序性问题的 Java 内存模型（JMM）。

什么是 Java 内存模型

Java 内存模型在维基百科上的定义如下：

The Java memory model describes how threads in the Java programming language interact through memory. Together with the description of single-threaded execution of code, the memory model provides the semantics of the Java programming language.

内存模型限制的是共享变量，也就是存储在堆内存中的变量，在 Java 语言中，所有的实例变量、静态变量和数组元素都存储在堆内存之中。而方法参数、异常处理参数这些局部变量存储在方法栈帧之中，因此不会在线程之间共享，不会受到内存模型影响，也不存在内存可见性问题。

通常，在线程之间的通讯方式有共享内存和消息传递两种，很明显，Java 采用的是第一种即共享的内存模型，在共享的内存模型里，多线程之间共享程序的公共状态，通过读-写内存的方式来进行隐式通讯。

从抽象的角度来看，JMM 其实是定义了线程和主内存之间的关系，首先，多个线程之间的共享变量存储在主内存之中，同时每个线程都有一个自己私有的本地内存，本地内存中存储着该线程读或写共享变量的副本（注意：本地内存是 JMM 定义的抽象概念，实际上并不存在）。抽象模型如下图所示：

在这个抽象的内存模型中，在两个线程之间的通信（共享变量状态变更）时，会进行如下两个步骤：

1. 线程 A 把在本地内存更新后的共享变量副本的值，刷新到主内存中。
2. 线程 B 在使用到该共享变量时，到主内存中去读取线程 A 更新后的共享变量的值，并更新线程 B 本地内存的值。

JMM 本质上是在硬件（处理器）内存模型之上又做了一层抽象，使得应用开发人员只需要了解 JMM 就可以编写出正确的并发代码，而无需过多了解硬件层面的内存模型。

为什么需要 Java 内存模型

在日常的程序开发中，为一些共享变量赋值的场景会经常碰到，假设一个线程为整型共享变量 count 做赋值操作（count = 9527;），此时就会有一个问题，其它读取该共享变量的线程在什么情况下获取到的变量值为 9527 呢？如果缺少同步的话，会有很多因素导致其它读取该变量的线程无法立即甚至是永远都无法看到该变量的最新值。

比如缓存就可能会改变写入共享变量副本提交到主内存的次序，保存在本地缓存的值，对于其它线程是不可见的；编译器为了优化性能，有时候会改变程序中语句执行的先后顺序，这些因素都有可能会导致其它线程无法看到共享变量的最新值。

在文章开头，提到了 JMM 主要是为了解决可见性和有序性问题，那么首先就要先搞清楚，导致可见性和有序性问题发生的本质原因是什么？现在的服务绝大部分都是运行在多核 CPU 的服务器上，每颗 CPU 都有自己的缓存，这时 CPU 缓存与内存的数据就会有一致性问题了，当一个线程对共享变量的修改，另外一个线程无法立刻看到。导致可见性问题的本质原因是缓存。

有序性是指代码实际的执行顺序和代码定义的顺序一致，编译器为了优化性能，虽然会遵守 as-if-serial 语义（不管怎么重排序，在单线程下的执行结果不能改变），不过有时候编译器及解释器的优化也可能引发一些问题。比如：双重检查来创建单实例对象。下面是使用双重检查来实现延迟创建单例对象的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

/**
 * @author mghio
 * @since 2021-08-22
 */
public class DoubleCheckedInstance {

  private static DoubleCheckedInstance instance;

  public static DoubleCheckedInstance getInstance() {
    if (instance == null) {
      synchronized (DoubleCheckedInstance.class) {
        if (instance == null) {
          instance = new DoubleCheckedInstance();
        }
      }
    }

    return instance;
  }
  
}

这里的 instance = new DoubleCheckedInstance();，看起来 Java 代码只有一行，应该是无法就行重排序的，实际上其编译后的实际指令是如下三步：

1. 分配对象的内存空间
2. 初始化对象
3. 设置 instance 指向刚刚已经分配的内存地址

上面的第 2 步和第 3 步如果改变执行顺序也不会改变单线程的执行结果，也就是说可能会发生重排序，下图是一种多线程并发执行的场景：

此时线程 B 获取到的 instance 是没有初始化过的，如果此来访问 instance 的成员变量就可能触发空指针异常。导致有序性问题的本质原因是编译器优化。那你可能会想既然缓存和编译器优化是导致可见性问题和有序性问题的原因，那直接禁用掉不就可以彻底解决这些问题了吗，但是如果这么做了的话，程序的性能可能就会受到比较大的影响了。

其实可以换一种思路，能不能把这些禁用缓存和编译器优化的权利交给编码的工程师来处理，他们肯定最清楚什么时候需要禁用，这样就只需要提供按需禁用缓存和编译优化的方法即可，使用比较灵活。因此Java 内存模型就诞生了，它规范了 JVM 如何提供按需禁用缓存和编译优化的方法，规定了 JVM 必须遵守一组最小的保证，这个最小保证规定了线程对共享变量的写入操作何时对其它线程可见。

顺序一致性内存模型

顺序一致性模型是一个理想化后的理论参考模型，处理器和编程语言的内存模型的设计都是参考的顺序一致性模型理论。其有如下两大特性：

1. 一个线程中的所有操作必须按照程序的顺序来执行
2. 所有的线程都只能看到一个单一的执行操作顺序，不管程序是否同步

在工程师视角下的顺序一致性模型如下：

顺序一致性模型有一个单一的全局内存，这个全局内存可以通过左右摇摆的开关可以连接到任意一个线程，每个线程都必须按照程序的顺序来执行内存的读和写操作。该理想模型下，任务时刻都只能有一个线程可以连接到内存，当多个线程并发执行时，就可以通过开关就可以把多个线程的读和写操作串行化。

顺序一致性模型中，所有操操作完全按照顺序串行执行，但是在 JMM 中就没有这个保证了，未同步的程序在 JMM 中不仅程序的执行顺序是无序的，而且由于本地内存的存在，所有线程看到的操作顺序也可能会不一致，比如一个线程把写共享变量保存在本地内存中，在还没有刷新到主内存前，其它线程是不可见的，只有更新到主内存后，其它线程才有可能看到。

JMM 对在正确同步的程序做了顺序一致性的保证，也就是程序的执行结果和该程序在顺序一致性内存模型中的执行结果相同。

Happens-Before 规则

Happens-Before 规则是 JMM 中的核心概念，Happens-Before 概念最开始在 这篇论文 提出，其在论文中使用 Happens-Before 来定义分布式系统之间的偏序关系。在 JSR-133 中使用 Happens-Before 来指定两个操作之间的执行顺序。

JMM 正是通过这个规则来保证跨线程的内存可见性，Happens-Before 的含义是前面一个对共享变量的操作结果对该变量的后续操作是可见的，约束了编译器的优化行为，虽然允许编译器优化，但是优化后的代码必须要满足 Happens-Before 规则，这个规则给工程师做了这个保证：同步的多线程程序是按照 Happens-Before 指定的顺序来执行的。目的就是为了在不改变程序（单线程或者正确同步的多线程程序）执行结果的前提下，尽最大可能的提高程序执行的效率。

JSR-133 规范中定了如下 6 项 Happens-Before 规则：

1. 程序顺序规则：一个线程中的每个操作，Happens-Before 该线程中的任意后续操作
2. 监视器锁规则：对一个锁的解锁操作，Happens-Before 于后面对这个锁的加锁操作
3. volatile 规则对一个 volatile 类型的变量的写操作，Happens-Before 与任意后面对这个 volatile 变量的读操作
4. 传递性规则：如果操作 A Happens-Before 于操作 B，并且操作 B Happens-Before 于操作 C，则操作 A Happens-Before 于操作 C
5. start() 规则：如果一个线程 A 执行操作 threadB.start() 启动线程 B，那么线程 A 的 start() 操作 Happens-Before 于线程 B 的任意操作
6. join() 规则：如果线程 A 执行操作 threadB.join() 并成功返回，那么线程 B 中的任意操作 Happens-Before 于线程 A 从 threadB.join() 操作成功返回

JMM 的一个基本原则是：只要不改变单线程和正确同步的多线程的执行结果，编译器和处理器随便怎么优化都可以，实际上对于应用开发人员对于两个操作是否真的被重排序并不关心，真正关心的是执行结果不能被修改。因此 Happens-Before 本质上和 sa-if-serial 的语义是一致的，只是 sa-if-serial 只是保证在单线程下的执行结果不被改变。

总结

本文主要介绍了内存模型的相关基础知识和相关概念，JMM 屏蔽了不同处理器内存模型之间的差异，在不同的处理器平台上给应用开发人员抽象出了统一的 Java 内存模型（JMM）。常见的处理器内存模型比 JMM 的要弱，因此 JVM 会在生成字节码指令时在适当的位置插入内存屏障（内存屏障的类型会因处理器平台而有所不同）来限制部分重排序。

前言

异步编程是让程序并发运行的一种手段。它允许多个事情同时发生，当程序调用需要长时间运行的方法时，它不会阻塞当前的执行流程，程序可以继续运行，当方法执行完成时通知给主线程根据需要获取其执行结果或者失败异常的原因。使用异步编程可以大大提高我们程序的吞吐量，可以更好的面对更高的并发场景并更好的利用现有的系统资源，同时也会一定程度上减少用户的等待时间等。本文我们一起来看看在 Java 语言中使用异步编程有哪些方式。

Thread 方式

在 Java 语言中最简单使用异步编程的方式就是创建一个 Thread 来实现，如果你使用的 JDK 版本是 8 以上的话，可以使用 Lambda 表达式 会更加简洁。为了能更好的体现出异步的高效性，下面提供同步版本和异步版本的示例作为对照：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

/**
 * @author mghio
 * @since 2021-08-01
 */
public class SyncWithAsyncDemo {

  public static void doOneThing() {
    try {
      Thread.sleep(2000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("doOneThing ---->>> success");
  }

  public static void doOtherThing() {
    try {
      Thread.sleep(2000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("doOtherThing ---->>> success");
  }

  public synchronized static void main(String[] args) throws InterruptedException {
    StopWatch stopWatch = new StopWatch("SyncWithAsyncDemo");
    stopWatch.start();

    // 同步调用版本
    // testSynchronize();

    // 异步调用版本
    testAsynchronize();

    stopWatch.stop();
    System.out.println(stopWatch);
  }

  private static void testAsynchronize() throws InterruptedException {
    System.out.println("-------------------- testAsynchronize --------------------");

    // 创建一个线程执行 doOneThing
    Thread doOneThingThread = new Thread(SyncWithAsyncDemo::doOneThing, "doOneThing-Thread");
    doOneThingThread.start();

    doOtherThing();
    // 等待 doOneThing 线程执行完成
    doOneThingThread.join();
  }

  private static void testSynchronize() {
    System.out.println("-------------------- testSynchronize --------------------");

    doOneThing();
    doOtherThing();
  }

}

同步执行的运行如下：

注释掉同步调用版本的代码，得到异步执行的结果如下：

从两次的运行结果可以看出，同步版本耗时 4002 ms，异步版本执行耗时 2064 ms，异步执行耗时减少将近一半，可以看出使用异步编程后可以大大缩短程序运行时间。

上面的示例的异步线程代码在 main 方法内开启了一个线程 doOneThing-Thread 用来异步执行 doOneThing 任务，在这时该线程与 main 主线程并发运行，也就是任务 doOneThing 与任务 doOtherThing 并发运行，则等主线程运行完 doOtherThing 任务后同步等待线程 doOneThing 运行完毕，整体还是比较简单的。

但是这个示例只能作为示例使用，如果用到了生产环境发生事故后果自负，使用上面这种 Thread 方式异步编程存在两个明显的问题。

1. 创建线程没有复用。我们知道频繁的线程创建与销毁是需要一部分开销的，而且示例里也没有限制线程的个数，如果使用不当可能会把系统线程用尽，从而引发事故，这个问题使用线程池可以解决。
2. 异步任务无法获取最终的执行结果。示例中的这种方式是满足不了的，这时候就需要使用下面介绍的第二种 FutureTask 的方式了。

FutureTask 方式

自 JDK 1.5 开始，引入了 Future 接口和实现 Future 接口的 FutureTask 类来表示异步计算结果。这个 FutureTask 类不仅实现了 Future 接口还实现了 Runnable 接口，表示一种可生成结果的 Runnable。其可以处于这三种状态：

• 未启动 当创建一个 FutureTask 没有执行 FutureTask.run() 方法之前
• 已启动 在 FutureTask.run() 方法执行的过程中
• 已完成 在 FutureTask.run() 方法正常执行结果或者调用了 FutureTask.cancel(boolean mayInterruptIfRunning) 方法以及在调用 FutureTask.run() 方法的过程中发生异常结束后

FutureTask 类实现了 Future 接口的开启和取消任务、查询任务是否完成、获取计算结果方法。要获取 FutureTask 任务的结果，我们只能通过调用 getXXX() 系列方法才能获取，当结果还没出来时候这些方法会被阻塞，同时这了任务可以是 Callable 类型（有返回结果），也可以是 Runnable 类型（无返回结果）。我们修改上面的示例把两个任务方法修改为返回 String 类型，使用 FutureTask 的方法如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

private static void testFutureTask() throws ExecutionException, InterruptedException {
    System.out.println("-------------------- testFutureTask --------------------");

    // 创建一个 FutureTask（doOneThing 任务）
    FutureTask<String> futureTask = new FutureTask<>(FutureTaskDemo::doOneThing);
    // 使用线程池执行 doOneThing 任务
    ForkJoinPool.commonPool().execute(futureTask);

    // 执行 doOtherThing 任务
    String doOtherThingResult = doOtherThing();

    // 同步等待线程执行 doOneThing 任务结束
    String doOneThingResult = futureTask.get();

    // 任务执行结果输出
    System.out.println("doOneThingResult ---->>> " + doOneThingResult);
    System.out.println("doOtherThingResult ---->>> " + doOtherThingResult);
}

使用 FutureTask 异步编程方式的耗时和上面的 Thread 方式是差不多的，其本质都是另起一个线程去做 doOneThing 任务然后等待返回，运行结果如下：

这个示例中，doOneThing 和 doOtherThing 都是有返回值的任务（都返回 String 类型结果），我们在主线程 main 中创建一个异步任务 FutureTask 来执行 doOneThing，然后使用 ForkJoinPool.commonPool() 创建线程池（有关 ForkJoinPool 的介绍见 这里），然后调用了线程池的 execute 方法把 futureTask 提交到线程池来执行。

通过示例可以看到，虽然 FutureTask 提供了一些方法让我们获取任务的执行结果、任务是否完成等，但是使用还是比较复杂，在一些较为复杂的场景（比如多个 FutureTask 之间的关系表示）的编码还是比较繁琐，还是当我们调用 getXXX() 系列方法时还是会在任务执行完毕前阻塞调用线程，达不到异步编程的效果，基于这些问题，在 JDK 8 中引入了 CompletableFuture 类，下面来看看如何使用 CompletableFuture 来实现异步编程。

CompletableFuture 方式

JDK 8 中引入了 CompletableFuture 类，实现了 Future 和 CompletionStage 接口，为异步编程提供了一些列方法，如 supplyAsync、runAsync 和 thenApplyAsync 等，除此之外 CompletableFuture 还有一个重要的功能就是可以让两个或者多个 CompletableFuture 进行运算来产生结果。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

/**
 * @author mghio
 * @since 2021-08-01
 */
public class CompletableFutureDemo {

  public static CompletableFuture<String> doOneThing() {
    return CompletableFuture.supplyAsync(() -> {
      try {
        Thread.sleep(2000);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      return "doOneThing";
    });
  }

  public static CompletableFuture<String> doOtherThing(String parameter) {
    return CompletableFuture.supplyAsync(() -> {
      try {
        Thread.sleep(2000);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      return parameter + " " + "doOtherThing";
    });
  }

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    StopWatch stopWatch = new StopWatch("CompletableFutureDemo");
    stopWatch.start();

    // 异步执行版本
    testCompletableFuture();

    stopWatch.stop();
    System.out.println(stopWatch);
  }

  private static void testCompletableFuture() throws InterruptedException, ExecutionException {
    // 先执行 doOneThing 任务，后执行 doOtherThing 任务
    CompletableFuture<String> resultFuture = doOneThing().thenCompose(CompletableFutureDemo::doOtherThing);

    // 获取任务结果
    String doOneThingResult = resultFuture.get();

    // 获取执行结果
    System.out.println("DoOneThing and DoOtherThing execute finished. result = " + doOneThingResult);
  }

}

执行结果如下：

在主线程 main 中首先调用了方法 doOneThing() 方法开启了一个异步任务，并返回了对应的 CompletableFuture 对象，我们取名为 doOneThingFuture，然后在 doOneThingFuture 的基础上使用 CompletableFuture 的 thenCompose() 方法，让 doOneThingFuture 方法执行完成后，使用其执行结果作为 doOtherThing(String parameter) 方法的参数创建的异步任务返回。

我们不需要显式使用 ExecutorService，在 CompletableFuture 内部使用的是 Fork/Join 框架异步处理任务，因此，它使我们编写的异步代码更加简洁。此外，CompletableFuture 类功能很强大其提供了和很多方便的方法，更多关于 CompletableFuture 的使用请见 这篇。

总结

本文介绍了在 Java 中的 JDK 使用异步编程的三种方式，这些是我们最基础的实现异步编程的工具，在其之上的还有 Guava 库提供的 ListenableFuture 和 Futures 类以及 Spring 框架提供的异步执行能力，使用 @Async 等注解实现异步处理，感兴趣的话可以自行学习了解。

什么是 Fork/Join 框架

Fork/Join 框架是一种在 JDK 7 引入的线程池，用于并行执行把一个大任务拆成多个小任务并行执行，最终汇总每个小任务结果得到大任务结果的特殊任务。通过其命名也很容易看出框架主要分为 Fork 和 Join 两个阶段，第一阶段 Fork 是把一个大任务拆分为多个子任务并行的执行，第二阶段 Join 是合并这些子任务的所有执行结果，最后得到大任务的结果。

这里不难发现其执行主要流程：首先判断一个任务是否足够小，如果任务足够小，则直接计算，否则，就拆分成几个更小的小任务分别计算，这个过程可以反复的拆分成一系列小任务。Fork/Join 框架是一种基于 分治 的算法，通过拆分大任务成多个独立的小任务，然后并行执行这些小任务，最后合并小任务的结果得到大任务的最终结果，通过并行计算以提高效率。

Fork/Join 框架使用示例

下面通过一个计算列表中所有元素的总和的示例来看看 Fork/Join 框架是如何使用的，总的思路是：将这个列表分成许多子列表，然后对每个子列表的元素进行求和，然后，我们再计算所有这些值的总和就得到原始列表的和了。Fork/Join 框架中定义了 ForkJoinTask 来表示一个 Fork/Join 任务，其提供了 fork()、join() 等操作，通常情况下，我们并不需要直接继承这个 ForkJoinTask 类，而是使用框架提供的两个 ForkJoinTask 的子类：

• RecursiveAction 用于表示没有返回结果的 Fork/Join 任务。
• RecursiveTask 用于表示有返回结果的 Fork/Join 任务。

很显然，在这个示例中是需要返回结果的，可以定义 SumAction 类继承自 RecursiveTask，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

/**
 * @author mghio
 * @since 2021-07-25
 */
public class SumTask extends RecursiveTask<Long> {

  private static final int SEQUENTIAL_THRESHOLD = 50;

  private final List<Long> data;

  public SumTask(List<Long> data) {
    this.data = data;
  }

  @Override
  protected Long compute() {
    if (data.size() <= SEQUENTIAL_THRESHOLD) {
      long sum = computeSumDirectly();
      System.out.format("Sum of %s: %d\n", data.toString(), sum);
      return sum;
    } else {
      int mid = data.size() / 2;
      SumTask firstSubtask = new SumTask(data.subList(0, mid));
      SumTask secondSubtask = new SumTask(data.subList(mid, data.size()));
      // 执行子任务
      firstSubtask.fork();
      secondSubtask.fork();
      // 等待子任务执行完成，并获取结果
      long firstSubTaskResult = firstSubtask.join();
      long secondSubTaskResult = secondSubtask.join();
      return firstSubTaskResult + secondSubTaskResult;
    }
  }

  private long computeSumDirectly() {
    long sum = 0;
    for (Long l : data) {
      sum += l;
    }
    return sum;
  }

  public static void main(String[] args) {
    Random random = new Random();

    List<Long> data = random
        .longs(1_000, 1, 100)
        .boxed()
        .collect(Collectors.toList());

    ForkJoinPool pool = new ForkJoinPool();
    SumTask task = new SumTask(data);
    pool.invoke(task);

    System.out.println("Sum: " + pool.invoke(task));
  }
}

这里当列表大小小于 SEQUENTIAL_THRESHOLD 变量的值（阈值）时视为小任务，直接计算求和列表元素结果，否则再次拆分为小任务，运行结果如下：

通过这个示例代码可以发现，Fork/Join 框架 中 ForkJoinTask 任务与平常的一般任务的主要不同点在于：ForkJoinTask 需要实现抽象方法 compute() 来定义计算逻辑，在这个方法里一般通用的实现模板是，首先先判断当前任务是否是小任务，如果是，就执行执行任务，如果不是小任务，则再次拆分为两个子任务，然后当每个子任务调用 fork() 方法时，会再次进入到 compute() 方法中，检查当前任务是否需要再拆分为子任务，如果已经是小任务，则执行当前任务并返回结果，否则继续分割，最后调用 join() 方法等待所有子任务执行完成并获得执行结果。伪代码如下：

1
2
3
4
5
6
7
8

if (problem is small) {
  directly solve problem.
} else {
  Step 1. split problem into independent parts.
  Step 2. fork new subtasks to solve each part.
  Step 3. join all subtasks.
  Step 4. compose result from subresults.
}

Fork/Join 框架设计

Fork/Join 框架核心思想是把一个大任务拆分成若干个小任务，然后汇总每个小任务的结果最终得到大任务的结果，如果让你设计一个这样的框架，你会如何实现呢？（建议思考一下），Fork/Join 框架的整个流程正如其名所示，分为两个步骤：

1. 大任务分割 需要有这么一个的类，用来将大任务拆分为子任务，可能一次拆分后的子任务还是比较大，需要多次拆分，直到拆分出来的子任务符合我们定义的小任务才结束。
2. 执行任务并合并任务结果 第一步拆分出来的子任务分别存放在一个个 双端队列 里面（P.S. 这里为什么要使用双端队列请看下文），然后每个队列启动一个线程从队列中获取任务执行。这些子任务的执行结果都会放到一个统一的队列中，然后再启动一个线程从这个队列中拿数据，最后合并这些数据返回。

Fork/Join 框架使用了如下两个类来完成以上两个步骤：

• ForkJoinTask 类 在上文的实例中也有提到，表示 ForkJoin 任务，在使用框架时首先必须先定义任务，通常只需要继承自 ForkJoinTask 类的子类 RecursiveAction(无返回结果) 或者 RecursiveTask(有返回结果)即可。
• ForkJoinPool 从名字也可以猜到一二了，就是用来执行 ForkJoinTask 的线程池。大任务拆分出的子任务会添加到当前线程的双端队列的头部。

喜欢思考的你，心中想必会想到这么一种场景，当我们需要完成一个大任务时，会先把这个大任务拆分为多个独立的子任务，这些子任务会放到独立的队列中，并为每个队列都创建一个单独的线程去执行队列里的任务，即这里线程和队列时一对一的关系，那么当有的线程可能会先把自己队列的任务执行完成了，而有的线程则没有执行完成，这就导致一些先执行完任务的线程干等了，这是个好问题。

既然是做并发的，肯定要最大程度压榨计算机的性能，对于这种场景并发大师 Doug Lea 使用了工作窃取算法处理，使用工作窃取算法后，先完成自己队列中任务的线程会去其它线程的队列中”窃取“一个任务来执行，哈哈，一方有难，八方支援。但是此时这个线程和队列的持有线程会同时访问同一个队列，所以为了减少窃取任务的线程和被窃取任务的线程之间的竞争，ForkJoin 选择了双端队列这种数据结构，这样就可以按照这种规则执行任务了：被窃取任务的线程始终从队列头部获取任务并执行，窃取任务的线程使用从队列尾部获取任务执行。这个算法在绝大部分情况下都可以充分利用多线程进行并行计算，但是在双端队列里只有一个任务等极端情况下还是会存在一定程度的竞争。

Fork/Join 框架实现原理

Fork/Join 框架的实现核心是 ForkJoinPool 类，该类的重要组成部分为 ForkJoinTask 数组和 ForkJoinWorkerThread 数组，其中 ForkJoinTask 数组用来存放框架使用者给提交给 ForkJoinPool 的任务，ForkJoinWorkerThread 数组则负责执行这些任务。任务有如下四种状态：

• NORMAL 已完成
• CANCELLED 被取消
• SIGNAL 信号
• EXCEPTIONAL 发生异常

下面来看看这两个类的核心方法实现原理，首先来看 ForkJoinTask 的 fork() 方法，源码如下：

方法对于 ForkJoinWorkerThread 类型的线程，首先会调用 ForkJoinWorkerThread 的 workQueue 的 push() 方法异步的去执行这个任务，然后马上返回结果。继续跟进 ForkJoinPool 的 push() 方法，源码如下：

方法将当前任务添加到 ForkJoinTask 任务队列数组中，然后再调用 ForkJoinPool 的 signalWork 方法创建或者唤醒一个工作线程来执行该任务。然后再来看看 ForkJoinTask 的 join() 方法，方法源码如下：

方法首先调用了 doJoin() 方法，该方法返回当前任务的状态，根据返回的任务状态做不同的处理：

1. 已完成状态则直接返回结果
2. 被取消状态则直接抛出异常（CancellationException）
3. 发生异常状态则直接抛出对应的异常

继续跟进 doJoin() 方法，方法源码如下：

方法首先判断当前任务状态是否已经执行完成，如果执行完成则直接返回任务状态。如果没有执行完成，则从任务数组中（workQueue）取出任务并执行，任务执行完成则设置任务状态为 NORMAL，如果出现异常则记录异常并设置任务状态为 EXCEPTIONAL（在 doExec() 方法中）。

总结

本文主要介绍了 Java 并发框架中的 Fork/Join 框架的基本原理和其使用的工作窃取算法(work-stealing)、设计方式和部分实现源码。Fork/Join 框架在 JDK 的官方标准库中也有应用。比如 JDK 1.8+ 标准库提供的 Arrays.parallelSort(array) 可以进行并行排序，它的原理就是内部通过 Fork/Join 框架对大数组分拆进行并行排序，可以提高排序的速度，还有集合中的 Collection.parallelStream() 方法底层也是基于 Fork/Join 框架实现的，最后就是定义小任务的阈值往往是需要通过测试验证才能合理给出，并且保证程序可以达到最好的性能。

前言

在 上篇 介绍了 Feign 的核心实现原理，在文末也提到了会再介绍其和 Spring Cloud 的整合原理，Spring 具有很强的扩展性，会把一些常用的解决方案通过 starter 的方式开放给开发者使用，在引入官方提供的 starter 后通常只需要添加一些注解即可使用相关功能（通常是 @EnableXXX）。下面就一起来看看 Spring Cloud 到底是如何整合 Feign 的。

整合原理浅析

在 Spring 中一切都是围绕 Bean 来展开的工作，而所有的 Bean 都是基于 BeanDefinition 来生成的，可以说 BeanDefinition 是整个 Spring 帝国的基石，这个整合的关键也就是要如何生成 Feign 对应的 BeanDefinition。

要分析其整合原理，我们首先要从哪里入手呢？如果你看过 上篇 的话，在介绍结合 Spring Cloud 使用方式的例子时，第二步就是要在项目的 XXXApplication 上加添加 @EnableFeignClients 注解，我们可以从这里作为切入点，一步步深入分析其实现原理（通常相当一部分的 starter 一般都是在启动类中添加了开启相关功能的注解）。

进入 @EnableFeignClients 注解中，其源码如下：

从注解的源码可以发现，该注解除了定义几个参数（basePackages、defaultConfiguration、clients 等）外，还通过 @Import 引入了 FeignClientsRegistrar 类，一般 @Import 注解有如下功能（具体功能可见 官方 Java Doc）：

• 声明一个 Bean
• 导入 @Configuration 注解的配置类
• 导入 ImportSelector 的实现类
• 导入 ImportBeanDefinitionRegistrar 的实现类（这里使用这个功能）

到这里不难看出，整合实现的主要流程就在 FeignClientsRegistrar 类中了，让我们继续深入到类 FeignClientsRegistrar 的源码，

通过源码可知 FeignClientsRegistrar 实现 ImportBeanDefinitionRegistrar 接口，该接口从名字也不难看出其主要功能就是将所需要初始化的 BeanDefinition 注入到容器中，接口定义两个方法功能都是用来注入给定的 BeanDefinition 的，一个可自定义 beanName（通过实现 BeanNameGenerator 接口自定义生成 beanName 的逻辑），另一个使用默认的规则生成 beanName（类名首字母小写格式）。接口源码如下所示：

对 Spring 有一些了解的朋友们都知道，Spring 会在容器启动的过程中根据 BeanDefinition 的属性信息完成对类的初始化，并注入到容器中。所以这里 FeignClientsRegistrar 的终极目标就是将生成的代理类注入到 Spring 容器中。
虽然 FeignClientsRegistrar 这个类的源码看起来比较多，但是从其终结目标来看，我们主要是看如何生成 BeanDefinition 的，通过源码可以发现其实现了 ImportBeanDefinitionRegistrar 接口，并且重写了 registerBeanDefinitions(AnnotationMetadata, BeanDefinitionRegistry) 方法，在这个方法里完成了一些 BeanDefinition 的生成和注册工作。源码如下：

整个过程主要分为如下两个步骤：

1. 给 @EnableFeignClients 的全局默认配置（注解的 defaultConfiguration 属性）创建 BeanDefinition 对象并注入到容器中（对应上图中的第 ① 步）
2. 给标有了 @FeignClient 的类创建 BeanDefinition 对象并注入到容器中（对应上图中的第 ② 步）

下面分别深入方法源码实现来看其具体实现原理，首先来看看第一步的方法 registerDefaultConfiguration(AnnotationMetadata, BeanDefinitionRegistry)，源码如下：

可以看到这里只是获取一下注解 @EnableFeignClients 的默认配置属性 defaultConfiguration 的值，最终的功能实现交给了 registerClientConfiguration(BeanDefinitionRegistry, Object, Object) 方法来完成，继续跟进深入该方法，其源码如下：

可以看到，全局默认配置的 BeanClazz 都是 FeignClientSpecification，然后这里将全局默认配置 configuration 设置为 BeanDefinition 构造器的输入参数，然后当调用构造器实例化时将这个参数传进去。到这里就已经把 @EnableFeignClients 的全局默认配置（注解的 defaultConfiguration 属性）创建出 BeanDefinition 对象并注入到容器中了，第一步到此完成，整体还是比较简单的。

下面再来看看第二步 给标有了 @FeignClient 的类创建 BeanDefinition 对象并注入到容器中 是如何实现的。深入第二步的方法 registerFeignClients(AnnotationMetadata, BeanDefinitionRegistry) 实现中，由于方法实现代码较多，使用截图会比较分散，所以用贴出源代码并在相关位置添加必要注释的方式进行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

public void registerFeignClients(AnnotationMetadata metadata, BeanDefinitionRegistry registry) {
    // 最终获取到有 @FeignClient 注解类的集合
    LinkedHashSet<BeanDefinition> candidateComponents = new LinkedHashSet<>();
    // 获取 @EnableFeignClients 注解的属性 map
    Map<String, Object> attrs = metadata.getAnnotationAttributes(EnableFeignClients.class.getName());
    // 获取 @EnableFeignClients 注解的 clients 属性
    final Class<?>[] clients = attrs == null ? null : (Class<?>[]) attrs.get("clients");
    if (clients == null || clients.length == 0) {  
        // 如果 @EnableFeignClients 注解未指定 clients 属性则扫描添加（扫描过滤条件为：标注有 @FeignClient 的类）
        ClassPathScanningCandidateComponentProvider scanner = getScanner();
        scanner.setResourceLoader(this.resourceLoader);
        scanner.addIncludeFilter(new AnnotationTypeFilter(FeignClient.class));
        Set<String> basePackages = getBasePackages(metadata);
        for (String basePackage : basePackages) {
            candidateComponents.addAll(scanner.findCandidateComponents(basePackage));
        }
    }
    else {
        // 如果 @EnableFeignClients 注解已指定 clients 属性，则直接添加，不再扫描（从这里可以看出，为了加快容器启动速度，建议都指定 clients 属性）
        for (Class<?> clazz : clients) {
            candidateComponents.add(new AnnotatedGenericBeanDefinition(clazz));
        }
    }

    // 遍历最终获取到的 @FeignClient 注解类的集合
    for (BeanDefinition candidateComponent : candidateComponents) {
        if (candidateComponent instanceof AnnotatedBeanDefinition) {
            // verify annotated class is an interface
            // 验证带注释的类必须是接口，不是接口则直接抛出异常（大家可以想一想为什么只能是接口？）
            AnnotatedBeanDefinition beanDefinition = (AnnotatedBeanDefinition) candidateComponent;
            AnnotationMetadata annotationMetadata = beanDefinition.getMetadata();
            Assert.isTrue(annotationMetadata.isInterface(), "@FeignClient can only be specified on an interface");
            // 获取 @FeignClient 注解的属性值
            Map<String, Object> attributes = annotationMetadata
                    .getAnnotationAttributes(FeignClient.class.getCanonicalName());
            // 获取 clientName 的值，也就是在构造器的参数值（具体获取逻辑可以参见 getClientName(Map<String, Object>) 方法      
            String name = getClientName(attributes);
            // 同上文第一步最后调用的方法，注入 @FeignClient 注解的配置对象到容器中
            registerClientConfiguration(registry, name, attributes.get("configuration"));
            // 注入 @FeignClient 对象，该对象可以在其它类中通过 @Autowired 直接引入（e.g. XXXService）
            registerFeignClient(registry, annotationMetadata, attributes);
        }
    }
}

通过源码可以看到最后是通过方法 registerFeignClient(BeanDefinitionRegistry, AnnotationMetadata, Map<String, Object>) 注入的 @FeignClient 对象，继续深入该方法，源码如下：

方法实现比较长，最终目标是构造出 BeanDefinition 对象，然后通过 BeanDefinitionReaderUtils.registerBeanDefinition(BeanDefinitionHolder, BeanDefinitionRegistry) 注入到容器中。

其中关键的一步是从 @FeignClient 注解中获取信息并设置到 BeanDefinitionBuilder 中，BeanDefinitionBuilder 中注册的类是 FeignClientFactoryBean，这个类的功能正如它的名字一样是用来创建出 FeignClient 的 Bean 的，然后 Spring 会根据 FeignClientFactoryBean 生成对象并注入到容器中。

需要明确的一点是，实际上这里最终注入到容器当中的是 FeignClientFactoryBean 这个类，Spring 会在类初始化的时候会根据这个类来生成实例对象，就是调用 FeignClientFactoryBean.getObject() 方法，这个生成的对象就是我们实际使用的代理对象。下面再进入到类 FeignClientFactoryBean 的 getObject() 这个⽅法，源码如下：

可以看到这个方法是直接调用的类中的另一个方法 getTarget() 的，在继续跟进该方法，由于该方法实现代码较多，使用截图会比较分散，所以用贴出源代码并在相关位置添加必要注释的方式进行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

/**
  * @param <T> the target type of the Feign client
  * @return a {@link Feign} client created with the specified data and the context
  * information
  */
<T> T getTarget() {
    // 从 Spring 容器中获取 FeignContext Bean
    FeignContext context = beanFactory != null ? beanFactory.getBean(FeignContext.class)
            : applicationContext.getBean(FeignContext.class);
    // 根据获取到的 FeignContext 构建出 Feign.Builder         
    Feign.Builder builder = feign(context);

    // 注解 @FeignClient 未指定 url 属性 
    if (!StringUtils.hasText(url)) {
        // url 属性是固定访问某一个实例地址，如果未指定协议则拼接 http 请求协议
        if (!name.startsWith("http")) {
            url = "http://" + name;
        }
        else {
            url = name;
        }
        // 格式化 url
        url += cleanPath();
        // 生成代理和我们之前的代理一样，注解 @FeignClient 未指定 url 属性则返回一个带有负载均衡功能的客户端对象
        return (T) loadBalance(builder, context, new HardCodedTarget<>(type, name, url));
    }
    // 注解 @FeignClient 已指定 url 属性 
    if (StringUtils.hasText(url) && !url.startsWith("http")) {
        url = "http://" + url;
    }
    String url = this.url + cleanPath();
    // 获取一个 client
    Client client = getOptional(context, Client.class);
    if (client != null) {
        if (client instanceof FeignBlockingLoadBalancerClient) {
            // not load balancing because we have a url,
            // but Spring Cloud LoadBalancer is on the classpath, so unwrap
            // 这里没有负载是因为我们有指定了 url 
            client = ((FeignBlockingLoadBalancerClient) client).getDelegate();
        }
        builder.client(client);
    }
    // 生成代理和我们之前的代理一样，最后被注入到 Spring 容器中
    Targeter targeter = get(context, Targeter.class);
    return (T) targeter.target(this, builder, context, new HardCodedTarget<>(type, name, url));
}

通过源码得知 FeignClientFactoryBean 继承了 FactoryBean，其方法 FactoryBean.getObject 返回的就是 Feign 的代理对象，最后这个代理对象被注入到 Spring 容器中，我们就通过 @Autowired 可以直接注入使用了。同时还可以发现上面的代码分支最终都会走到如下代码：

1
2

Targeter targeter = get(context, Targeter.class);
return targeter.target(this, builder, context, target);

点进去深入 targeter.target 的源码，可以看到实际上这里创建的就是一个代理对象，也就是说在容器启动的时候，会为每个 @FeignClient 创建了一个代理对象。至此，Spring Cloud 和 Feign 整合原理的核心实现介绍完毕。

总结

本文主要介绍了 Spring Cloud 整合 Feign 的原理。通过上文介绍，你已经知道 Srpring 会我们的标注的 @FeignClient 的接口创建了一个代理对象，那么有了这个代理对象我们就可以做增强处理（e.g. 前置增强、后置增强），那么你知道是如何实现的吗？感兴趣的朋友可以再翻翻源码寻找答案（温馨提示：增强逻辑在 InvocationHandler 中）。还有 Feign 与 Ribbon 和 Hystrix 等组件的协作，感兴趣的朋友可以自行下载源码学习了解。

搜索能力是被绝大多数人低估一项基本素质，绝大部分做编程技术相关的朋友应该都知道如何使用 Google，但是并不知道如何利用它的潜力。其实不管是 Google 还是 百度，会搜索的人一样都可以查找到需要的东西，不会搜索的人用什么都不好使。下面介绍一些 Google 常用的搜索技巧以及搜索快捷方式，可以帮助你更快，更准确地找到结果。Google 是世界上功能最强大的搜索引擎，它已经改变了我们查找信息的方式。

0. 使用准确的词组

将您要搜索的关键字用引号引起来，Google 会进行精确的词组搜索。

语法：”[searchkey 1] [searchkey 2]” [searchkey 3]

1. 多个互斥的搜索条件使用 OR

默认情况下，除非指定，否则 Google 会包含你搜索条件中的所有搜索关键字。通过在您的关键词之字输入OR，Google 会知道它可以查找一组或另一组。大写 OR，否则 Google 会认为它只是你的关键字的一部分。

语法：[searchkey 1] OR [searchkey 2]

2. 排除指定关键词

通过在单词的前面添加减号，将单词从 Google 搜索中排除。

语法：-[searchkey to exclude] [searchkey to include]

3. 查找文本块中的所有单词

使用 Google 的 allintext： 语法仅搜索网站的正文，而忽略链接，URL 和标题。

语法：allintext:[searchkeys]

4. 在文本 + 标题 + URL 等查找单词

查找搜索词在不同位置的网页。即-在页面正文中，页面标题，URL 等中。为此，在您的关键字之前使用 intext:。

语法：intext:[searchkeys]

5. 标题搜索（单个关键字）

在网页标题内搜索一个单词，然后在网页上的其他位置搜索另一个单词。为此，您需要将 intitle： 混合到您的搜索查询中。

语法：[searchkeys 1] intitle:[searchkeys 2]

6. 标题搜索（多个关键字）

在网页标题中搜索查询中的所有关键字，在我们的搜索词之前使用 allintitle：。

语法：allintitle:[searchkey1 searchkey2]

7. 在 URL 中搜索

使用 allinURL 可以很容易地在 URL 中搜索关键字 。

语法：allinURL:[searchkeys]

8. 在指定网站内搜索

在网站内搜索单词-使用网站 URL 前面的 site： 语法，后跟您的搜索词。这会将搜索结果仅限制在该网站上。

语法：site:[website URL] [searchkeys]

9. Google 搜索定义

通过在单词之前使用 define： 轻松地找到单词的定义，而无需访问词典网站。Google 将提供定义，并提供一个音频播放器来提供该单词的语音发音。

语法：define:[searchkey]

10. Google 搜索通配符（遗漏或未知词）

没想到所有的话吗？加上 * 告诉 Google 为您填写空白，这对于歌曲歌词或书名搜索非常有效。

语法：[searchkeys 1] * [searchkeys 2]

11. Google 搜索文件类型

搜索文件类型（例如 PowerPoint，PDF 等）时，请在搜索词中使用 filetype： 命令。

语法：[searchkeyword] filetype:[file type extension]

12. 转换计算

使用 Google 可以进行任何度量转换。

语法：convert [data value + unit of measure] to [like unit of measure]

13. Google 搜索计算器

在搜索栏中输入您的计算结果，将 Google 用作计算器。数值运算符: * 表示乘，+ 表示加，- 表示减，/ 表示除。

语法：[number] [operator] [number]

14. Google 图片搜索

查找图像的名称，描述和类型。

语法：[searchkeyw] image type