AQS源码分析之Elasticsearch BaseFuture

多线程模式

Java多线程编程中，常用的多线程设计模式包括：Future模式、Master-Worker模式、Guarded Suspeionsion模式、不变模式和生产者-消费者模式等

Future模式

Future模式对于多线程而言，如果线程A要等待线程B的结果，那么线程A没必要一直等待B，直到B有结果，可以先拿到一个未来的Future，等B有结果时再获取真实的结果。Future模式的核心在于：去除了主函数的等待时间，并使得原本需要等待的时间段可以用于处理其他业务逻辑。

Future模式有点类似于商品订单。在网上购物时，提交订单后，在收货的这段时间里无需一直在家里等候，可以先干别的事情。类推到程序设计中时，当提交请求时，期望得到答复时，如果这个答复可能很慢。传统的是一直持续等待直到这个答复收到之后再去做别的事情，但如果利用Future模式，其调用方式改为异步，而原先等待返回的时间段，在主调用函数中，则可以用于处理其他事务。

Future模式的应用

Future模式的应用场景非常多，比如在rxjava，guava，dubbo的client与server交互的response与request也是用future进行包装的，还有很多其他的场景中也都离不开Future的身影。下面我们顺便来看下Future在Elasticsearch中的应用。future的主要有以下几个方法：

• cancel(boolean) ：尝试取消任务的执行；
• get()：尝试获取任务执行的结果，是一个阻塞方法；
• get(long,TimeUnit)：获取任务执行结果的方法，阻塞到指定时间；
• isDone()：任务是否执行完成；
• isCancelled()：任务是否已经取消。

org.elasticsearch.common.util.concurrent.BaseFuture

我们先来分析一下它的内部类org.elasticsearch.common.util.concurrent.BaseFuture.Sync：

1. /**
2. 1. 遵循{@link AbstractQueuedSynchronizer}的约定，我们创建了一个私有子类来保存同步器。该同步器用于实现阻塞和等待调用以及以线程安全的方式处理状态更改。将来的当前状态保持为“同步”状态，并且只要状态更改为{@link #COMPLETED}或{@link #CANCELLED}，就会释放锁定。
3. 2. 为了避免进行释放和获取的线程之间的竞争，我们分两步过渡到最终状态。一个线程将成功地将CAS从RUNNING转换为COMPLETING，然后该线程将设置计算结果，然后才转换为COMPLETED或CANCELLED。
4. 3. 我们不使用在aqs中acquire方法之间传递的整数参数，因此我们在各处传递-1来填充这个参数。
5. */
6. static final class Sync<V> extends AbstractQueuedSynchronizer {
7. /* Valid states. */
8. static final int RUNNING = 0;// 初始状态，AQS的state默认为0
9. static final int COMPLETING = 1;
10. static final int COMPLETED = 2;
11. static final int CANCELLED = 4;
13. private V value;
14. private Throwable exception;
16. /*
17. * Acquisition succeeds if the future is done, otherwise it fails.
18. */
19. @Override
20. protected int tryAcquireShared(int ignored) {//当future完成时获取成功，否则获取失败
21. if (isDone()) {// future完成
22. return 1;
23. }
24. return -1;
25. }
27. /*
28. * We always allow a release to go through, this means the state has been
29. * successfully changed and the result is available.
30. */
31. @Override
32. protected boolean tryReleaseShared(int finalState) {// 释放
33. // 设置最终状态
34. setState(finalState);
35. return true;
36. }
38. /**
39. * Blocks until the task is complete or the timeout expires. Throws a
40. * {@link TimeoutException} if the timer expires, otherwise behaves like
41. * {@link #get()}.
42. */
43. V get(long nanos) throws TimeoutException, CancellationException,
44. ExecutionException, InterruptedException {//阻塞直到任务完成或者时间超时
46. // Attempt to acquire the shared lock with a timeout.
47. /**
48. 这里调用的是java.util.concurrent.locks.AbstractQueuedSynchronizer#tryAcquireSharedNanos方法
49. public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
50. throws InterruptedException {
51. if (Thread.interrupted())
52. throw new InterruptedException();
53. return tryAcquireShared(arg) >= 0 ||
54. doAcquireSharedNanos(arg, nanosTimeout);
55. }
56. 1. 这里调用的是上面的tryAcquireShared方法，在任务完成时返回1，否则返回-1，当返回1时会直接返回true，不会执行doAcquireSharedNanos方法；
57. 2. 如果任务没有完成，则tryAcquireShared(arg) >= 0为false，那么会进入doAcquireSharedNanos方法，在该方法中会在未超时的时间内调用tryAcquireShared方法再次尝试，也是在任务执行完成时返回true，否则进行阻塞直到超时释放并返回false。当两个结果都为false时会进入下面这个判断。
58. */
59. if (!tryAcquireSharedNanos(-1, nanos)) {
60. throw new TimeoutException("Timeout waiting for task.");
61. }
62. // 如果能进行到这里，证明任务已经执行完成，可以获取到结果
63. return getValue();
64. }
66. /**
67. * Blocks until {@link #complete(Object, Throwable, int)} has been
68. * successfully called. Throws a {@link CancellationException} if the task
69. * was cancelled, or a {@link ExecutionException} if the task completed with
70. * an error.
71. */
72. V get() throws CancellationException, ExecutionException,
73. InterruptedException {
74. /**
75. 这里我们看下方法：
76. public final void acquireSharedInterruptibly(int arg)
77. throws InterruptedException {
78. if (Thread.interrupted())
79. throw new InterruptedException();
80. if (tryAcquireShared(arg) < 0)
81. doAcquireSharedInterruptibly(arg);
82. }
83. 1. tryAcquireShared方法在任务未完成时会返回-1，然后进入doAcquireSharedInterruptibly方法;
84. 2. 再来看doAcquireSharedInterruptibly方法，里面会调用tryAcquireShared方法进行判断，如果任务完成则返回值为1，会直接返回，否则代表任务未完成，线程会被park，进行等待状态，直到complete方法被调用。
85. */
86. // Acquire the shared lock allowing interruption.
87. acquireSharedInterruptibly(-1);
88. return getValue();
89. }
91. /**
92. * Implementation of the actual value retrieval. Will return the value
93. * on success, an exception on failure, a cancellation on cancellation, or
94. * an illegal state if the synchronizer is in an invalid state.
95. */
96. private V getValue() throws CancellationException, ExecutionException {
97. int state = getState();
98. switch (state) {
99. case COMPLETED:
100. if (exception != null) {
101. throw new ExecutionException(exception);
102. } else {
103. // 返回执行结果
104. return value;
105. }
107. case CANCELLED:
108. throw new CancellationException("Task was cancelled.");
110. default:
111. throw new IllegalStateException(
112. "Error, synchronizer in invalid state: " + state);
113. }
114. }
116. /**
117. * Checks if the state is {@link #COMPLETED} or {@link #CANCELLED}.
118. */
119. boolean isDone() {
120. // 是否执行完成
121. return (getState() & (COMPLETED | CANCELLED)) != 0;
122. }
124. /**
125. * Checks if the state is {@link #CANCELLED}.
126. */
127. boolean isCancelled() {
128. // 是否处于取消状态
129. return getState() == CANCELLED;
130. }
132. /**
133. * Transition to the COMPLETED state and set the value.
134. */
135. boolean set(@Nullable V v) {//过渡到COMPLETED状态并设置值
136. // 设置状态
137. return complete(v, null, COMPLETED);
138. }
140. /**
141. * Transition to the COMPLETED state and set the exception.
142. */
143. boolean setException(Throwable t) {// 过渡到COMPLETED状态并设置异常
144. return complete(null, t, COMPLETED);
145. }
147. /**
148. * Transition to the CANCELLED state.
149. */
150. boolean cancel() {//过渡到CANCELLED状态
151. return complete(null, null, CANCELLED);
152. }
154. /**
155. * Implementation of completing a task. Either {@code v} or {@code t} will
156. * be set but not both. The {@code finalState} is the state to change to
157. * from {@link #RUNNING}. If the state is not in the RUNNING state we
158. * return {@code false} after waiting for the state to be set to a valid
159. * final state ({@link #COMPLETED} or {@link #CANCELLED}).
160. *
161. * @param v the value to set as the result of the computation.
162. * @param t the exception to set as the result of the computation.
163. * @param finalState the state to transition to.
164. */
165. private boolean complete(@Nullable V v, @Nullable Throwable t,
166. int finalState) {
167. // cas AQS的state从初始的RUNNING到COMPLETING
168. boolean doCompletion = compareAndSetState(RUNNING, COMPLETING);
169. if (doCompletion) {//如果cas成功了
170. // If this thread successfully transitioned to COMPLETING, set the value
171. // and exception and then release to the final state.
172. this.value = v;//设置结果
173. this.exception = t;// 设置异常
174. /**
175. public final boolean releaseShared(int arg) {
176. if (tryReleaseShared(arg)) {
177. doReleaseShared();// 在里面执行线程唤醒的任务，unparkSuccessor
178. return true;
179. }
180. return false;
181. }
182. */
183. releaseShared(finalState);// 设置最终状态
184. } else if (getState() == COMPLETING) {// 如果是完成中
185. // If some other thread is currently completing the future, block until
186. // they are done so we can guarantee completion.
187. acquireShared(-1);// 如果有线程处于完成中，阻塞直到他们全部完成
188. }
189. // 返回结果
190. return doCompletion;
191. }
192. }

• 遵循{@link AbstractQueuedSynchronizer}的约定，我们创建了一个私有子类来保存同步器。该同步器用于实现阻塞和等待调用以及以线程安全的方式处理状态更改。将来的当前状态保持为“同步”状态，并且只要状态更改为{@link #COMPLETED}或{@link #CANCELLED}，就会释放锁定。
• 为了避免进行释放和获取的线程之间的竞争，我们分两步过渡到最终状态。一个线程将成功地将CAS从RUNNING转换为COMPLETING，然后该线程将设置计算结果，然后才转换为COMPLETED或CANCELLED。
• 我们不使用在aqs中acquire方法之间传递的整数参数，因此我们在各处传递-1来填充这个参数。
• get(long nanos)方法，get()方法和complete方法都是可能会阻塞的，这些可以理解成都是属于client方法，即外层的方法，具体细节参考代码注释内容。
• 唤醒操作依赖于 set(@Nullable V v)、setException(Throwable t)、cancel()等方法的调用，因为在这些方法中都调用了complete()方法并进行了state的更新。这些可以理解成处理任务的server端的方法，即内层通知外层任务执行完成的方法。
• 有一点需要注意，这里可能会存在外层多个线程同时get的情况，如果此时任务已经完成，则结果立即返回，否则多个线程都会进入阻塞，然后在complete方法的releaseShared方法中进行unpark。

BaseFuture中的属性与方法一览：

1. public abstract class BaseFuture<V> implements Future<V> {
3. private static final String BLOCKING_OP_REASON = "Blocking operation";
5. /**
6. * Synchronization control for AbstractFutures.
7. */
8. private final Sync<V> sync = new Sync<>();
10. @Override
11. public V get(long timeout, TimeUnit unit) throws InterruptedException,
12. TimeoutException, ExecutionException {
13. assert timeout <= 0 || blockingAllowed();
14. // 这里调用的是sync的get（timeout）方法
15. return sync.get(unit.toNanos(timeout));
16. }
18. /**
19. * {@inheritDoc}
20. * <p>
21. * The default {@link BaseFuture} implementation throws {@code
22. * InterruptedException} if the current thread is interrupted before or during
23. * the call, even if the value is already available.
24. *
25. * @throws InterruptedException if the current thread was interrupted before
26. * or during the call (optional but recommended).
27. * @throws CancellationException {@inheritDoc}
28. */
29. @Override
30. public V get(long timeout, TimeUnit unit) throws InterruptedException,
31. TimeoutException, ExecutionException {
32. assert timeout <= 0 || blockingAllowed();
33. // 调用的是sync的get(timeout)方法
34. return sync.get(unit.toNanos(timeout));
35. }
37. /*
38. * Improve the documentation of when InterruptedException is thrown. Our
39. * behavior matches the JDK's, but the JDK's documentation is misleading.
40. */
42. /**
43. * {@inheritDoc}
44. * <p>
45. * The default {@link BaseFuture} implementation throws {@code
46. * InterruptedException} if the current thread is interrupted before or during
47. * the call, even if the value is already available.
48. *
49. * @throws InterruptedException if the current thread was interrupted before
50. * or during the call (optional but recommended).
51. * @throws CancellationException {@inheritDoc}
52. */
53. @Override
54. public V get() throws InterruptedException, ExecutionException {
55. assert blockingAllowed();
56. return sync.get();
57. }
59. // protected so that it can be overridden in specific instances
60. protected boolean blockingAllowed() {// 用于es集群状态、处理线程等的校验，可以被使用的地方重写
61. return Transports.assertNotTransportThread(BLOCKING_OP_REASON) &&
62. ThreadPool.assertNotScheduleThread(BLOCKING_OP_REASON) &&
63. ClusterApplierService.assertNotClusterStateUpdateThread(BLOCKING_OP_REASON) &&
64. MasterService.assertNotMasterUpdateThread(BLOCKING_OP_REASON);
65. }
67. @Override
68. public boolean isDone() {
69. return sync.isDone();
70. }
72. @Override
73. public boolean isCancelled() {
74. return sync.isCancelled();
75. }
77. @Override
78. public boolean cancel(boolean mayInterruptIfRunning) {
79. if (!sync.cancel()) {
80. return false;
81. }
82. done();
83. if (mayInterruptIfRunning) {
84. interruptTask();
85. }
86. return true;
87. }
89. // 模板方法
90. protected void interruptTask() {
91. }
93. /**
94. * Subclasses should invoke this method to set the result of the computation
95. * to {@code value}. This will set the state of the future to
96. * {@link BaseFuture.Sync#COMPLETED} and call {@link #done()} if the
97. * state was successfully changed.
98. *
99. * @param value the value that was the result of the task.
100. * @return true if the state was successfully changed.
101. */
102. protected boolean set(@Nullable V value) {
103. boolean result = sync.set(value);
104. if (result) {
105. done();
106. }
107. return result;
108. }

可以看到BaseFuture中的方法大都依赖于Sync来实现的。

总结

全篇设计的最妙的地方是用Sync包装了Future的方法，然后对tryAcquireShared方法和tryReleaseShared方法的重写也是整个设计的核心。以上纯属个人观点，不当之处还请指正。