Eureka 源码解析 —— 任务批处理
总阅读量:4783次
摘要: 原创出处 http://www.iocoder.cn/Eureka/batch-tasks/ 「芋道源码」欢迎转载，保留摘要，谢谢！
本文主要基于 Eureka 1.8.X 版本
- 1. 概述
- 2. 整体流程
- 3. 任务处理器
- 4. 创建任务分发器
- 5. 创建任务接收执行器
- 6. 创建任务执行器
- 7. 网络通信整形器
- 8. 任务接收执行器【处理任务】
- 9. 任务接收线程【调度任务】
- 10. 任务执行器【执行任务】
- 666. 彩蛋
1. 概述
本文主要分享 任务批处理。Eureka-Server 集群通过任务批处理同步应用实例注册实例，所以本文也是为 Eureka-Server 集群同步的分享做铺垫。
本文涉及类在 com.netflix.eureka.util.batcher 包下，涉及到主体类的类图如下( 打开大图 )：
- 紫色部分 —— 任务分发器
- 蓝色部分 —— 任务接收器
- 红色部分 —— 任务执行器
- 绿色部分 —— 任务处理器
- 黄色部分 —— 任务持有者( 任务 )
推荐 Spring Cloud 书籍：
- 请支持正版。下载盗版，等于主动编写低级 BUG 。
- 程序猿DD —— 《Spring Cloud微服务实战》
- 周立 —— 《Spring Cloud与Docker微服务架构实战》
- 两书齐买，京东包邮。
2. 整体流程
任务执行的整体流程如下( 打开大图 )：
-
细箭头 —— 任务执行经历的操作
-
粗箭头 —— 任务队列流转的方向
-
不同于一般情况下，任务提交了立即同步或异步执行，任务的执行拆分了三层队列：
-
第一层，接收队列(
acceptorQueue)，重新处理队列(reprocessQueue)。- 蓝线：分发器在收到任务执行请求后，提交到接收队列，任务实际未执行。
- 黄线：执行器的工作线程处理任务失败，将符合条件( 见 「3. 任务处理器」 )的失败任务提交到重新执行队列。
-
第二层，待执行队列(
processingOrder)- 粉线：接收线程( Runner )将重新执行队列，接收队列提交到待执行队列。
-
第三层，工作队列(
workQueue)- 粉线：接收线程( Runner )将待执行队列的任务根据参数(
maxBatchingSize)将任务合并成批量任务，调度( 提交 )到工作队列。 - 黄线：执行器的工作线程池，一个工作线程可以拉取一个批量任务进行执行。
- 粉线：接收线程( Runner )将待执行队列的任务根据参数(
-
-
三层队列的好处：
- 接收队列，避免处理任务的阻塞等待。
- 接收线程( Runner )合并任务，将相同任务编号( 是的，任务是带有编号的 )的任务合并，只执行一次。
- Eureka-Server 为集群同步提供批量操作多个应用实例的接口，一个批量任务可以一次调度接口完成，避免多次调用的开销。当然，这样做的前提是合并任务，这也导致 Eureka-Server 集群之间对应用实例的注册和下线带来更大的延迟。毕竟，Eureka 是在 CAP 之间，选择了 AP。
3. 任务处理器
com.netflix.eureka.util.batcher.TaskProcessor ，任务处理器接口。接口代码如下：
public interface TaskProcessor<T> { |
- ProcessingResult ，处理任务结果。
Success，成功。Congestion，拥挤错误，任务将会被重试。例如，请求被限流。TransientError，瞬时错误，任务将会被重试。例如，网络请求超时。PermanentError，永久错误，任务将会被丢弃。例如，执行时发生程序异常。
#process(task)方法，处理单任务。#process(tasks)方法，处理批量任务。
4. 创建任务分发器
com.netflix.eureka.util.batcher.TaskDispatcher ，任务分发器接口。接口代码如下：
public interface TaskDispatcher<ID, T> { |
#process(...)方法，提交任务编号，任务，任务过期时间给任务分发器处理。
com.netflix.eureka.util.batcher.TaskDispatchers ，任务分发器工厂类，用于创建任务分发器。其内部提供两种任务分发器的实现：
- 批量任务执行的分发器，用于 Eureka-Server 集群注册信息的同步任务。
- 单任务执行的分发器，用于 Eureka-Server 向亚马逊 AWS 的 ASG ( Autoscaling Group ) 同步状态。虽然本系列暂时对 AWS 相关的不做解析，从工具类的角度来说，本文会对该分发器进行分享。
com.netflix.eureka.cluster.ReplicationTaskProcessor ，实现 TaskDispatcher ，Eureka-Server 集群任务处理器。感兴趣的同学，可以点击链接自己研究，我们将在 《Eureka 源码解析 —— Eureka-Server 集群同步》 有详细解析。
4.1 批量任务执行分发器
调用 TaskDispatchers#createBatchingTaskDispatcher(...) 方法，创建批量任务执行的分发器，实现代码如下：
// TaskDispatchers.java |
- 第 1 至 23 行 ：方法参数。比较多哈，请耐心理解。
workloadSize参数，单个批量任务包含任务最大数量。taskProcessor参数，自定义任务执行器实现。
- 第 24 至 27 行 ：创建任务接收执行器。在 「5. 创建任务接收器」 详细解析。
- 第 28 至 29 行 ：创建批量任务执行器。在 「6.1 创建批量任务执行器」 详细解析。
- 第 30 至 42 行 ：创建批量任务分发器。
- 第 32 至 35 行 ：
#process()方法的实现，调用AcceptorExecutor#process(...)方法，提交 [ 任务编号 , 任务 , 任务过期时间 ] 给任务分发器处理。
- 第 32 至 35 行 ：
4.2 单任务执行分发器
调用 TaskDispatchers#createNonBatchingTaskDispatcher(...) 方法，创建单任务执行的分发器，实现代码如下：
1: /** |
- 第 1 至 21 行 ：方法参数。比较多哈，请耐心理解。
workloadSize参数#createBatchingTaskDispatcher(...)少这个参数。在第 24 行，你会发现该参数传递给 AcceptorExecutor 使用 1 噢。taskProcessor参数，自定义任务执行器实现。
- 第 21 至 25 行 ：创建任务接收执行器。和
#createBatchingTaskDispatcher(...)只差workloadSize = 1参数。在 「5. 创建任务接收器」 详细解析。 - 第 28 至 29 行 ：创建单任务执行器。和
#createBatchingTaskDispatcher(...)差别很大。「6.2 创建单任务执行器」 详细解析。 - 第 30 至 42 行 ：创建单任务分发器。和
#createBatchingTaskDispatcher(...)一样。
5. 创建任务接收执行器
com.netflix.eureka.util.batcher.AcceptorExecutor ，任务接收执行器。创建构造方法代码如下：
1: class AcceptorExecutor<ID, T> { |
- 第 5 至 61 行 ：属性。比较多哈，请耐心理解。
- 眼尖如你，会发现 AcceptorExecutor 即存在单任务工作队列(
singleItemWorkQueue)，又存在批量任务工作队列(batchWorkQueue) ，在 「9. 任务接收线程【调度任务】」 会解答这个疑惑。
- 眼尖如你，会发现 AcceptorExecutor 即存在单任务工作队列(
- 第 78 至 79 行 ：创建网络通信整形器。在 「7. 网络通信整形器」 详细解析。
- 第 81 至 85 行 ：创建接收任务线程。
6. 创建任务执行器
com.netflix.eureka.util.batcher.TaskExecutors ，任务执行器。其内部提供创建单任务和批量任务执行器的两种方法。TaskExecutors 构造方法如下：
class TaskExecutors<ID, T> { |
workerThreads属性，工作线程池。工作任务队列会被工作线程池并发拉取，并发执行。com.netflix.eureka.util.batcher.TaskExecutors.WorkerRunnableFactory，创建工作线程工厂接口。单任务和批量任务执行器的工作线程实现不同，通过自定义工厂实现类创建。
6.1 创建批量任务执行器
调用 TaskExecutors#batchExecutors(...) 方法，创建批量任务执行器。实现代码如下：
/** |
com.netflix.eureka.util.batcher.TaskExecutors.WorkerRunnable.BatchWorkerRunnable，批量任务工作线程。
6.2 创建单任务执行器
调用 TaskExecutors#singleItemExecutors(...) 方法，创建批量任务执行器。实现代码如下：
/** |
com.netflix.eureka.util.batcher.TaskExecutors.WorkerRunnable.SingleTaskWorkerRunnable，单任务工作线程。
6.3 工作线程抽象类
com.netflix.eureka.util.batcher.TaskExecutors.WorkerRunnable ，任务工作线程抽象类。BatchWorkerRunnable 和 SingleTaskWorkerRunnable 都实现该类，差异在 #run() 的自定义实现。WorkerRunnable 实现代码如下：
abstract static class WorkerRunnable<ID, T> implements Runnable { |
7. 网络通信整形器
com.netflix.eureka.util.batcher.TrafficShaper ，网络通信整形器。当任务执行发生请求限流，或是请求网络失败的情况，则延时 AcceptorRunner 将任务提交到工作任务队列，从而避免任务很快去执行，再次发生上述情况。TrafficShaper 实现代码如下：
class TrafficShaper { |
#registerFailure(...)，在任务执行失败时，提交任务结果给 TrafficShaper ，记录发生时间。在 「10. 任务执行器【执行任务】」 会看到调用该方法。#transmissionDelay(...)，计算提交延迟，单位：毫秒。「9. 任务接收线程【调度任务】」 会看到调用该方法。
8. 任务接收执行器【处理任务】
调用 AcceptorExecutor#process(...) 方法，添加任务到接收任务队列。实现代码如下：
// AcceptorExecutor.java |
-
com.netflix.eureka.util.batcher.TaskHolder，任务持有者，实现代码如下：class TaskHolder<ID, T> {
/**
* 任务编号
*/
private final ID id;
/**
* 任务
*/
private final T task;
/**
* 任务过期时间戳
*/
private final long expiryTime;
/**
* 任务提交时间戳
*/
private final long submitTimestamp;
}
9. 任务接收线程【调度任务】
后台线程执行 AcceptorRunner#run(...) 方法，调度任务。实现代码如下：
1: |
-
第 4 行 ：无限循环执行调度，直到关闭。
-
第 6 至 7 行 ：调用
#drainInputQueues()方法，循环处理完输入队列( 接收队列 + 重新执行队列 )，直到有待执行的任务。实现代码如下：1: private void drainInputQueues() throws InterruptedException {
2: do {
3: // 处理完重新执行队列
4: drainReprocessQueue();
5: // 处理完接收队列
6: drainAcceptorQueue();
7:
8: // 所有队列为空，等待 10 ms，看接收队列是否有新任务
9: if (!isShutdown.get()) {
10: // If all queues are empty, block for a while on the acceptor queue
11: if (reprocessQueue.isEmpty() && acceptorQueue.isEmpty() && pendingTasks.isEmpty()) {
12: TaskHolder<ID, T> taskHolder = acceptorQueue.poll(10, TimeUnit.MILLISECONDS);
13: if (taskHolder != null) {
14: appendTaskHolder(taskHolder);
15: }
16: }
17: }
18: } while (!reprocessQueue.isEmpty() || !acceptorQueue.isEmpty() || pendingTasks.isEmpty()); // 处理完输入队列( 接收队列 + 重新执行队列 )
19: }-
第 2 行 && 第 18 行 ：循环，直到同时满足如下全部条件：
- 重新执行队列(
reprocessQueue) 和接收队列(acceptorQueue)为空 - 待执行任务映射(
pendingTasks)不为空
- 重新执行队列(
-
第 3 至 4 行 ：处理完重新执行队列(
reprocessQueue)。实现代码如下：1: private void drainReprocessQueue() {
2: long now = System.currentTimeMillis();
3: while (!reprocessQueue.isEmpty() && !isFull()) {
4: TaskHolder<ID, T> taskHolder = reprocessQueue.pollLast(); // 优先拿较新的任务
5: ID id = taskHolder.getId();
6: if (taskHolder.getExpiryTime() <= now) { // 过期
7: expiredTasks++;
8: } else if (pendingTasks.containsKey(id)) { // 已存在
9: overriddenTasks++;
10: } else {
11: pendingTasks.put(id, taskHolder);
12: processingOrder.addFirst(id); // 提交到队头
13: }
14: }
15: // 如果待执行队列已满，清空重新执行队列，放弃较早的任务
16: if (isFull()) {
17: queueOverflows += reprocessQueue.size();
18: reprocessQueue.clear();
19: }
20: }- 第 4 行 ：优先从重新执行任务的队尾拿较新的任务，从而实现保留更新的任务在待执行任务映射(
pendingTasks) 里。 - 第 12 行 ：添加任务编号到待执行队列(
processingOrder) 的头部。效果如下图：
- 第 15 至 18 行 ：如果待执行队列(
pendingTasks)已满，清空重新执行队列(processingOrder)，放弃较早的任务。
- 第 4 行 ：优先从重新执行任务的队尾拿较新的任务，从而实现保留更新的任务在待执行任务映射(
-
第 5 至 6 行 ：处理完接收队列(
acceptorQueue)，实现代码如下：private void drainAcceptorQueue() {
while (!acceptorQueue.isEmpty()) { // 循环，直到接收队列为空
appendTaskHolder(acceptorQueue.poll());
}
}
private void appendTaskHolder(TaskHolder<ID, T> taskHolder) {
// 如果待执行队列已满，移除待处理队列，放弃较早的任务
if (isFull()) {
pendingTasks.remove(processingOrder.poll());
queueOverflows++;
}
// 添加到待执行队列
TaskHolder<ID, T> previousTask = pendingTasks.put(taskHolder.getId(), taskHolder);
if (previousTask == null) {
processingOrder.add(taskHolder.getId());
} else {
overriddenTasks++;
}
} -
第 8 至 17 行 ：当所有队列为空，阻塞从接收队列(
acceptorQueue) 拉取任务 10 ms。若拉取到，添加到待执行队列(processingOrder)。
-
-
第 12 至 16 行 ：计算可调度任务的最小时间(
scheduleTime)。- 当
scheduleTime小于当前时间，不重新计算，即此时需要延迟等待调度。 - 当
scheduleTime大于等于当前时间，配合TrafficShaper#transmissionDelay(...)重新计算。
- 当
-
第 19 行 ：当
scheduleTime小于当前时间，执行任务的调度。 -
第 21 行 ：调用
#assignBatchWork()方法，调度批量任务。实现代码如下：1: void assignBatchWork() {
2: if (hasEnoughTasksForNextBatch()) {
3: // 获取 批量任务工作请求信号量
4: if (batchWorkRequests.tryAcquire(1)) {
5: // 获取批量任务
6: long now = System.currentTimeMillis();
7: int len = Math.min(maxBatchingSize, processingOrder.size());
8: List<TaskHolder<ID, T>> holders = new ArrayList<>(len);
9: while (holders.size() < len && !processingOrder.isEmpty()) {
10: ID id = processingOrder.poll();
11: TaskHolder<ID, T> holder = pendingTasks.remove(id);
12: if (holder.getExpiryTime() > now) { // 过期
13: holders.add(holder);
14: } else {
15: expiredTasks++;
16: }
17: }
18: //
19: if (holders.isEmpty()) { // 未调度到批量任务，释放请求信号量
20: batchWorkRequests.release();
21: } else { // 添加批量任务到批量任务工作队列
22: batchSizeMetric.record(holders.size(), TimeUnit.MILLISECONDS);
23: batchWorkQueue.add(holders);
24: }
25: }
26: }
27: }-
第 2 行 ：调用
#hasEnoughTasksForNextBatch()方法，判断是否有足够任务进行下一次批量任务调度：1）待执行任务(processingOrder)映射已满；或者 2）到达批量任务处理最大等待延迟。实现代码如下：private boolean hasEnoughTasksForNextBatch() {
// 待执行队列为空
if (processingOrder.isEmpty()) {
return false;
}
// 待执行任务映射已满
if (pendingTasks.size() >= maxBufferSize) {
return true;
}
// 到达批量任务处理最大等待延迟( 通过待处理队列的头部任务判断 )
TaskHolder<ID, T> nextHolder = pendingTasks.get(processingOrder.peek());
long delay = System.currentTimeMillis() - nextHolder.getSubmitTimestamp();
return delay >= maxBatchingDelay;
}- x
-
第 5 至 17 行 ：获取批量任务(
holders)。😈 你会发现，本文说了半天的批量任务，实际是List<TaskHolder<ID, T>>哈。 -
第 4 行 ：获取批量任务工作请求信号量(
batchWorkRequests) 。在任务执行器的批量任务执行器，每次执行时，发出batchWorkRequests。每一个信号量需要保证获取到一个批量任务。 -
第 19 至 20 行 ：未调度到批量任务，释放请求信号量，代表请求实际未完成，每一个信号量需要保证获取到一个批量任务。
-
第 21 至 24 行 ：添加批量任务到批量任务工作队列。
-
第 23 行 ：调用
#assignSingleItemWork()方法，调度单任务。
-
-
第 23 行 ：调用
#assignSingleItemWork()方法，调度单任务，和#assignBatchWork()方法类似。实现代码如下：void assignSingleItemWork() {
if (!processingOrder.isEmpty()) { // 待执行任队列不为空
// 获取 单任务工作请求信号量
if (singleItemWorkRequests.tryAcquire(1)) {
// 【循环】获取单任务
long now = System.currentTimeMillis();
while (!processingOrder.isEmpty()) {
ID id = processingOrder.poll(); // 一定不为空
TaskHolder<ID, T> holder = pendingTasks.remove(id);
if (holder.getExpiryTime() > now) {
singleItemWorkQueue.add(holder);
return;
}
expiredTasks++;
}
// 获取不到单任务，释放请求信号量
singleItemWorkRequests.release();
}
}
}- x
-
第 26 至 31 行 ：当调度任务前的待执行任务数(
totalItems)等于当前待执行队列(processingOrder)的任务数，意味着：1）任务执行器无任务请求，正在忙碌处理之前的任务；或者 2）任务延迟调度。睡眠 10 秒，避免资源浪费。
10. 任务执行器【执行任务】
10.1 批量任务工作线程
批量任务工作后台线程( BatchWorkerRunnable )执行 #run(...) 方法，调度任务。实现代码如下：
// |
-
第 4 行 ：无限循环执行调度，直到关闭。
-
第 6 行 ：调用
getWork()方法，获取一个批量任务直到成功。实现代码如下：1: private List<TaskHolder<ID, T>> getWork() throws InterruptedException {
2: // 发起请求信号量，并获得批量任务的工作队列
3: BlockingQueue<List<TaskHolder<ID, T>>> workQueue = taskDispatcher.requestWorkItems();
4: // 【循环】获取批量任务，直到成功
5: List<TaskHolder<ID, T>> result;
6: do {
7: result = workQueue.poll(1, TimeUnit.SECONDS);
8: } while (!isShutdown.get() && result == null);
9: return result;
10: }-
第 3 行 ：调用
TaskDispatcher#requestWorkItems()方法，发起请求信号量，并获得批量任务的工作队列。实现代码如下：// TaskDispatcher.java
/**
* 批量任务工作请求信号量
*/
private final Semaphore batchWorkRequests = new Semaphore(0);
/**
* 批量任务工作队列
*/
private final BlockingQueue<List<TaskHolder<ID, T>>> batchWorkQueue = new LinkedBlockingQueue<>();
BlockingQueue<List<TaskHolder<ID, T>>> requestWorkItems() {
batchWorkRequests.release();
return batchWorkQueue;
}- 注意，批量任务工作队列(
batchWorkQueue) 和单任务工作队列(singleItemWorkQueue) 是不同的队列。
- 注意，批量任务工作队列(
-
第 5 至 8 行 ：循环获取一个批量任务，直到成功。
-
-
第 12 行 ：调用
#getTasksOf(...)方法，获得实际批量任务。实现代码如下：private List<T> getTasksOf(List<TaskHolder<ID, T>> holders) {
List<T> tasks = new ArrayList<>(holders.size());
for (TaskHolder<ID, T> holder : holders) {
tasks.add(holder.getTask());
}
return tasks;
}- x
-
第 14 至 24 行 ：调用处理器( TaskProcessor ) 执行任务。当任务执行结果为
Congestion或TransientError，调用AcceptorExecutor#reprocess(...)提交整个批量任务重新处理，实现代码如下：// AcceptorExecutor.java
void reprocess(List<TaskHolder<ID, T>> holders, ProcessingResult processingResult) {
// 添加到 重新执行队列
reprocessQueue.addAll(holders);
// TODO 芋艿：监控相关，暂时无视
replayedTasks += holders.size();
// 提交任务结果给 TrafficShaper
trafficShaper.registerFailure(processingResult);
}
10.2 单任务工作线程
单任务工作后台线程( SingleTaskWorkerRunnable )执行 #run(...) 方法，调度任务，和 BatchWorkerRunnable#run(...) 基本类似，就不啰嗦了。实现代码如下：
|
666. 彩蛋
😈 又是一篇长文。建议边看代码，边对照着整体流程图，理解实际不难。
当然，欢迎你有任何疑问，在我的公众号( 芋道源码 ) 留言。
胖友，分享我的公众号( 芋道源码 ) 给你的胖友可好？