Eureka 源码解析 —— 任务批处理

public interface TaskProcessor<T>{

 /**
* A processed task/task list ends up in one of the following states:
* <ul>
* <li>{@code Success} processing finished successfully</li>
* <li>{@code TransientError} processing failed, but shall be retried later</li>
* <li>{@code PermanentError} processing failed, and is non recoverable</li>
* </ul>
*/
 enum ProcessingResult {
 /**
* 成功
*/
 Success,
 /**
* 拥挤错误
*/
 Congestion,
 /**
* 瞬时错误
*/
 TransientError,
 /**
* 永久错误
*/
 PermanentError
 }

 /**
* 处理单任务
* In non-batched mode a single task is processed at a time.
*/
 ProcessingResult process(T task);

 /**
* 处理批量任务
*
* For batched mode a collection of tasks is run at a time. The result is provided for the aggregated result,
* and all tasks are handled in the same way according to what is returned (for example are rescheduled, if the
* error is transient).
*/
 ProcessingResult process(List<T> tasks);
}

public interface TaskDispatcher<ID, T>{

 void process(ID id, T task, long expiryTime);

 void shutdown();
}

// TaskDispatchers.java
 1: /**
2: * 创建批量任务执行的分发器
3: *
4: * @param id 任务执行器编号
5: * @param maxBufferSize 待执行队列最大数量
6: * @param workloadSize 单个批量任务包含任务最大数量
7: * @param workerCount 任务执行器工作线程数
8: * @param maxBatchingDelay 批量任务等待最大延迟时长，单位：毫秒
9: * @param congestionRetryDelayMs 请求限流延迟重试时间，单位：毫秒
10: * @param networkFailureRetryMs 网络失败延迟重试时长，单位：毫秒
11: * @param taskProcessor 任务处理器
12: * @param <ID> 任务编号泛型
13: * @param <T> 任务泛型
14: * @return 批量任务执行的分发器
15: */
 16: public static <ID, T> TaskDispatcher<ID, T> createBatchingTaskDispatcher(String id,
17: int maxBufferSize,
18: int workloadSize,
19: int workerCount,
20: long maxBatchingDelay,
21: long congestionRetryDelayMs,
22: long networkFailureRetryMs,
23: TaskProcessor<T> taskProcessor){
 24: // 创建 任务接收执行器
 25: final AcceptorExecutor<ID, T> acceptorExecutor = new AcceptorExecutor<>(
 26: id, maxBufferSize, workloadSize, maxBatchingDelay, congestionRetryDelayMs, networkFailureRetryMs
 27: );
 28: // 创建 批量任务执行器
 29: final TaskExecutors<ID, T> taskExecutor = TaskExecutors.batchExecutors(id, workerCount, taskProcessor, acceptorExecutor);
 30: // 创建 批量任务分发器
 31: return new TaskDispatcher<ID, T>() {
 32: @Override
 33: public void process(ID id, T task, long expiryTime){
 34: acceptorExecutor.process(id, task, expiryTime);
 35: }
 36: 
 37: @Override
 38: public void shutdown(){
 39: acceptorExecutor.shutdown();
 40: taskExecutor.shutdown();
 41: }
 42: };
 43: }

 1: /**
2: * 创建单任务执行的分发器
3: *
4: * @param id 任务执行器编号
5: * @param maxBufferSize 待执行队列最大数量
6: * @param workerCount 任务执行器工作线程数
7: * @param maxBatchingDelay 批量任务等待最大延迟时长，单位：毫秒
8: * @param congestionRetryDelayMs 请求限流延迟重试时间，单位：毫秒
9: * @param networkFailureRetryMs 网络失败延迟重试时长，单位：毫秒
10: * @param taskProcessor 任务处理器
11: * @param <ID> 任务编号泛型
12: * @param <T> 任务泛型
13: * @return 单任务执行的分发器
14: */
15: public static <ID, T> TaskDispatcher<ID, T> createNonBatchingTaskDispatcher(String id,
16: int maxBufferSize,
17: int workerCount,
18: long maxBatchingDelay,
19: long congestionRetryDelayMs,
20: long networkFailureRetryMs,
21: TaskProcessor<T> taskProcessor){
22: // 创建 任务接收执行器
23: final AcceptorExecutor<ID, T> acceptorExecutor = new AcceptorExecutor<>(
24: id, maxBufferSize, /* workloadSize = 1 */1, maxBatchingDelay, congestionRetryDelayMs, networkFailureRetryMs
25: );
26: final TaskExecutors<ID, T> taskExecutor = TaskExecutors.singleItemExecutors(id, workerCount, taskProcessor, acceptorExecutor);
27: return new TaskDispatcher<ID, T>() {
28: @Override
29: public void process(ID id, T task, long expiryTime){
30: acceptorExecutor.process(id, task, expiryTime);
31: }
32: 
33: @Override
34: public void shutdown(){
35: acceptorExecutor.shutdown();
36: taskExecutor.shutdown();
37: }
38: };
39: }

 1: class AcceptorExecutor<ID, T>{
 2: 
 3: private static final Logger logger = LoggerFactory.getLogger(AcceptorExecutor.class);
 4: 
 5: /**
6: * 待执行队列最大数量
7: * {@link #processingOrder}
8: */
 9: private final int maxBufferSize;
10: /**
11: * 单个批量任务包含任务最大数量
12: */
13: private final int maxBatchingSize;
14: /**
15: * 批量任务等待最大延迟时长，单位：毫秒
16: */
17: private final long maxBatchingDelay;
18: 
19: /**
20: * 是否关闭
21: */
22: private final AtomicBoolean isShutdown = new AtomicBoolean(false);
23: /**
24: * 接收任务队列
25: */
26: private final BlockingQueue<TaskHolder<ID, T>> acceptorQueue = new LinkedBlockingQueue<>();
27: /**
28: * 重新执行任务队列
29: */
30: private final BlockingDeque<TaskHolder<ID, T>> reprocessQueue = new LinkedBlockingDeque<>();
31: /**
32: * 接收任务线程
33: */
34: private final Thread acceptorThread;
35: 
36: /**
37: * 待执行任务映射
38: */
39: private final Map<ID, TaskHolder<ID, T>> pendingTasks = new HashMap<>();
40: /**
41: * 待执行队列
42: */
43: private final Deque<ID> processingOrder = new LinkedList<>();
44: 
45: /**
46: * 单任务工作请求信号量
47: */
48: private final Semaphore singleItemWorkRequests = new Semaphore(0);
49: /**
50: * 单任务工作队列
51: */
52: private final BlockingQueue<TaskHolder<ID, T>> singleItemWorkQueue = new LinkedBlockingQueue<>();
53: 
54: /**
55: * 批量任务工作请求信号量
56: */
57: private final Semaphore batchWorkRequests = new Semaphore(0);
58: /**
59: * 批量任务工作队列
60: */
61: private final BlockingQueue<List<TaskHolder<ID, T>>> batchWorkQueue = new LinkedBlockingQueue<>();
62: 
63: /**
64: * 网络通信整形器
65: */
66: private final TrafficShaper trafficShaper;
67: 
68: AcceptorExecutor(String id,
69: int maxBufferSize,
70: int maxBatchingSize,
71: long maxBatchingDelay,
72: long congestionRetryDelayMs,
73: long networkFailureRetryMs) {
74: this.maxBufferSize = maxBufferSize;
75: this.maxBatchingSize = maxBatchingSize;
76: this.maxBatchingDelay = maxBatchingDelay;
77: 
78: // 创建 网络通信整形器
79: this.trafficShaper = new TrafficShaper(congestionRetryDelayMs, networkFailureRetryMs);
80: 
81: // 创建 接收任务线程
82: ThreadGroup threadGroup = new ThreadGroup("eurekaTaskExecutors");
83: this.acceptorThread = new Thread(threadGroup, new AcceptorRunner(), "TaskAcceptor-" + id);
84: this.acceptorThread.setDaemon(true);
85: this.acceptorThread.start();
86: 
87: // TODO （省略代码）芋艿：监控相关，暂时无视
88: }
89: }

class TaskExecutors<ID, T>{

 private static final Logger logger = LoggerFactory.getLogger(TaskExecutors.class);

 /**
* 是否关闭
*/
 private final AtomicBoolean isShutdown;
 /**
* 工作线程池
*/
 private final List<Thread> workerThreads;

 TaskExecutors(WorkerRunnableFactory<ID, T> workerRunnableFactory, int workerCount, AtomicBoolean isShutdown) {
 this.isShutdown = isShutdown;
 this.workerThreads = new ArrayList<>();

 // 创建 工作线程池
 ThreadGroup threadGroup = new ThreadGroup("eurekaTaskExecutors");
 for (int i = 0; i < workerCount; i++) {
 WorkerRunnable<ID, T> runnable = workerRunnableFactory.create(i);
 Thread workerThread = new Thread(threadGroup, runnable, runnable.getWorkerName());
 workerThreads.add(workerThread);
 workerThread.setDaemon(true);
 workerThread.start();
 }
 }
 
 /**
* 创建工作线程工厂
*
* @param <ID> 任务编号泛型
* @param <T> 批量任务执行器
*/
 interface WorkerRunnableFactory<ID, T>{
 WorkerRunnable<ID, T> create(int idx);
 }
}

/**
* 创建批量任务执行器
*
* @param name 任务执行器名
* @param workerCount 任务执行器工作线程数
* @param processor 任务处理器
* @param acceptorExecutor 接收任务执行器
* @param <ID> 任务编号泛型
* @param <T> 任务泛型
* @return 批量任务执行器
*/
static <ID, T> TaskExecutors<ID, T> batchExecutors(final String name,
int workerCount,
final TaskProcessor<T> processor,
final AcceptorExecutor<ID, T> acceptorExecutor){
 final AtomicBoolean isShutdown = new AtomicBoolean();
 final TaskExecutorMetrics metrics = new TaskExecutorMetrics(name);
 // 创建批量任务执行器
 return new TaskExecutors<>(new WorkerRunnableFactory<ID, T>() { // 批量任务工作线程工厂
 @Override
 public WorkerRunnable<ID, T> create(int idx){
 return new BatchWorkerRunnable<>("TaskBatchingWorker-" + name + '-' + idx /* 线程名 */, isShutdown, metrics, processor, acceptorExecutor);
 }
 }, workerCount, isShutdown);
}

/**
* 创建单任务执行器
*
* @param name 任务执行器名
* @param workerCount 任务执行器工作线程数
* @param processor 任务处理器
* @param acceptorExecutor 接收任务执行器
* @param <ID> 任务编号泛型
* @param <T> 任务泛型
* @return 单任务执行器
*/
static <ID, T> TaskExecutors<ID, T> singleItemExecutors(final String name,
int workerCount,
final TaskProcessor<T> processor,
final AcceptorExecutor<ID, T> acceptorExecutor){
 final AtomicBoolean isShutdown = new AtomicBoolean();
 final TaskExecutorMetrics metrics = new TaskExecutorMetrics(name);
 // 创建单任务执行器
 return new TaskExecutors<>(new WorkerRunnableFactory<ID, T>() { // 单任务工作线程工厂
 @Override
 public WorkerRunnable<ID, T> create(int idx){
 return new SingleTaskWorkerRunnable<>("TaskNonBatchingWorker-" + name + '-' + idx /* 线程名 */, isShutdown, metrics, processor, acceptorExecutor);
 }
 }, workerCount, isShutdown);
}

abstract static class WorkerRunnable<ID, T> implements Runnable{
 /**
* 线程名
*/
 final String workerName;
 /**
* 是否关闭
*/
 final AtomicBoolean isShutdown;
 final TaskExecutorMetrics metrics;
 /**
* 任务处理器
*/
 final TaskProcessor<T> processor;
 /**
* 任务接收执行器
*/
 final AcceptorExecutor<ID, T> taskDispatcher;
 
 // ... 省略构造方法和 getting 方法。
}

class TrafficShaper{

 /**
* Upper bound on delay provided by configuration.
*/
 private static final long MAX_DELAY = 30 * 1000;

 /**
* 请求限流延迟重试时间，单位：毫秒
*/
 private final long congestionRetryDelayMs;
 /**
* 网络失败延迟重试时长，单位：毫秒
*/
 private final long networkFailureRetryMs;

 /**
* 最后请求限流时间戳，单位：毫秒
*/
 private volatile long lastCongestionError;
 /**
* 最后网络失败时间戳，单位：毫秒
*/
 private volatile long lastNetworkFailure;

 TrafficShaper(long congestionRetryDelayMs, long networkFailureRetryMs) {
 this.congestionRetryDelayMs = Math.min(MAX_DELAY, congestionRetryDelayMs);
 this.networkFailureRetryMs = Math.min(MAX_DELAY, networkFailureRetryMs);
 }

 void registerFailure(ProcessingResult processingResult){
 if (processingResult == ProcessingResult.Congestion) {
 lastCongestionError = System.currentTimeMillis();
 } else if (processingResult == ProcessingResult.TransientError) {
 lastNetworkFailure = System.currentTimeMillis();
 }
 }

 /**
* 计算提交延迟，单位：毫秒
*
* @return 延迟
*/
 long transmissionDelay(){
 // 无延迟
 if (lastCongestionError == -1 && lastNetworkFailure == -1) {
 return 0;
 }

 long now = System.currentTimeMillis();

 // 计算最后请求限流带来的延迟
 if (lastCongestionError != -1) {
 long congestionDelay = now - lastCongestionError;
 if (congestionDelay >= 0 && congestionDelay < congestionRetryDelayMs) { // 范围内
 return congestionRetryDelayMs - congestionDelay; // 补充延迟
 }
 lastCongestionError = -1; // 重置时间戳
 }

 // 计算最后网络失败带来的延迟
 if (lastNetworkFailure != -1) {
 long failureDelay = now - lastNetworkFailure;
 if (failureDelay >= 0 && failureDelay < networkFailureRetryMs) { // 范围内
 return networkFailureRetryMs - failureDelay; // 补充延迟
 }
 lastNetworkFailure = -1; // 重置时间戳
 }

 // 无延迟
 return 0;
 }
}

// AcceptorExecutor.java
void process(ID id, T task, long expiryTime){
 acceptorQueue.add(new TaskHolder<ID, T>(id, task, expiryTime));
 acceptedTasks++;
}

 1: @Override
 2: public void run(){
 3: long scheduleTime = 0;
 4: while (!isShutdown.get()) {
 5: try {
 6: // 处理完输入队列( 接收队列 + 重新执行队列 )
 7: drainInputQueues();
 8: 
 9: // 待执行任务数量
10: int totalItems = processingOrder.size();
11: 
12: // 计算调度时间
13: long now = System.currentTimeMillis();
14: if (scheduleTime < now) {
15: scheduleTime = now + trafficShaper.transmissionDelay();
16: }
17: 
18: // 调度
19: if (scheduleTime <= now) {
20: // 调度批量任务
21: assignBatchWork();
22: // 调度单任务
23: assignSingleItemWork();
24: }
25: 
26: // 1）任务执行器无任务请求，正在忙碌处理之前的任务；或者 2）任务延迟调度。睡眠 10 秒，避免资源浪费。
27: // If no worker is requesting data or there is a delay injected by the traffic shaper,
28: // sleep for some time to avoid tight loop.
29: if (totalItems == processingOrder.size()) {
30: Thread.sleep(10);
31: }
32: } catch (InterruptedException ex) {
33: // Ignore
34: } catch (Throwable e) {
35: // Safe-guard, so we never exit this loop in an uncontrolled way.
36: logger.warn("Discovery AcceptorThread error", e);
37: }
38: }
39: }

//
 1: @Override
 2: public void run(){
 3: try {
 4: while (!isShutdown.get()) {
 5: // 获取批量任务
 6: List<TaskHolder<ID, T>> holders = getWork();
 7: 
 8: // TODO 芋艿：监控相关，暂时无视
 9: metrics.registerExpiryTimes(holders);
 10: 
 11: // 获得实际批量任务
 12: List<T> tasks = getTasksOf(holders);
 13: // 调用处理器执行任务
 14: ProcessingResult result = processor.process(tasks);
 15: switch (result) {
 16: case Success:
 17: break;
 18: case Congestion:
 19: case TransientError:
 20: taskDispatcher.reprocess(holders, result); // 提交重新处理
 21: break;
 22: case PermanentError:
 23: logger.warn("Discarding {} tasks of {} due to permanent error", holders.size(), workerName);
 24: }
 25: 
 26: // TODO 芋艿：监控相关，暂时无视
 27: metrics.registerTaskResult(result, tasks.size());
 28: }
 29: } catch (InterruptedException e) {
 30: // Ignore
 31: } catch (Throwable e) {
 32: // Safe-guard, so we never exit this loop in an uncontrolled way.
 33: logger.warn("Discovery WorkerThread error", e);
 34: }
 35: }

@Override
// SingleTaskWorkerRunnable.java
public void run(){
 try {
 while (!isShutdown.get()) {
 // 发起请求信号量，并获得单任务的工作队列
 BlockingQueue<TaskHolder<ID, T>> workQueue = taskDispatcher.requestWorkItem();
 TaskHolder<ID, T> taskHolder;
 // 【循环】获取单任务，直到成功
 while ((taskHolder = workQueue.poll(1, TimeUnit.SECONDS)) == null) {
 if (isShutdown.get()) {
 return;
 }
 }
 // TODO 芋艿：监控相关，暂时无视
 metrics.registerExpiryTime(taskHolder);
 if (taskHolder != null) {
 // 调用处理器执行任务
 ProcessingResult result = processor.process(taskHolder.getTask());
 switch (result) {
 case Success:
 break;
 case Congestion:
 case TransientError:
 taskDispatcher.reprocess(taskHolder, result); // 提交重新处理
 break;
 case PermanentError:
 logger.warn("Discarding a task of {} due to permanent error", workerName);
 }
 // TODO 芋艿：监控相关，暂时无视
 metrics.registerTaskResult(result, 1);
 }
 }
 } catch (InterruptedException e) {
 // Ignore
 } catch (Throwable e) {
 // Safe-guard, so we never exit this loop in an uncontrolled way.
 logger.warn("Discovery WorkerThread error", e);
 }
}