Java 21虚拟线程与Project Reactor构建亿级异步任务处理系统实战指南

三、核心异步任务处理框架实现

3.1 任务定义与元数据

// Task.java - 任务基类
public abstract class Task<T> {
    private final String taskId;
    private final TaskPriority priority;
    private final Instant createdAt;
    private final Duration timeout;
    private final Map<String, Object> metadata;
    
    public enum TaskPriority {
        HIGH, MEDIUM, LOW, BATCH
    }
    
    public Task(TaskPriority priority, Duration timeout) {
        this.taskId = UUID.randomUUID().toString();
        this.priority = priority;
        this.createdAt = Instant.now();
        this.timeout = timeout;
        this.metadata = new ConcurrentHashMap();
    }
    
    // 抽象方法 - 任务执行逻辑
    public abstract Mono<T> execute();
    
    // 任务重试策略
    public RetrySpec getRetrySpec() {
        return Retry.backoff(3, Duration.ofSeconds(1))
                  .filter(this::isRetryable)
                  .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> 
                      new TaskExecutionException("Task execution failed after retries"));
    }
    
    protected boolean isRetryable(Throwable throwable) {
        return !(throwable instanceof BusinessException);
    }
    
    // 虚拟线程执行包装
    public Mono<T> executeWithVirtualThread() {
        return Mono.fromCallable(() -> {
            try (var scope = new StructuredTaskScope<T>()) {
                Future<T> future = scope.fork(this::executeBlocking);
                scope.join();
                return future.resultNow();
            }
        }).subscribeOn(Schedulers.fromExecutor(virtualThreadExecutor));
    }
    
    protected abstract T executeBlocking() throws Exception;
}

// 具体任务实现示例
public class PaymentProcessingTask extends Task<PaymentResult> {
    private final PaymentRequest request;
    
    public PaymentProcessingTask(PaymentRequest request) {
        super(TaskPriority.HIGH, Duration.ofSeconds(30));
        this.request = request;
    }
    
    @Override
    public Mono<PaymentResult> execute() {
        return Mono.fromCallable(() -> processPayment(request))
                  .timeout(getTimeout())
                  .retryWhen(getRetrySpec())
                  .metrics()
                  .name("payment.task.execution")
                  .tag("priority", getPriority().name());
    }
    
    private PaymentResult processPayment(PaymentRequest request) {
        // 支付处理逻辑
        return paymentService.process(request);
    }
    
    @Override
    protected PaymentResult executeBlocking() throws Exception {
        return processPayment(request);
    }
}

3.2 响应式任务调度器

// ReactiveTaskScheduler.java
@Component
@Slf4j
public class ReactiveTaskScheduler {
    
    private final Scheduler virtualThreadScheduler;
    private final MeterRegistry meterRegistry;
    private final ConcurrentHashMap<String, Disposable> scheduledTasks;
    
    public ReactiveTaskScheduler(MeterRegistry meterRegistry) {
        this.virtualThreadScheduler = Schedulers.fromExecutor(
            Executors.newVirtualThreadPerTaskExecutor()
        );
        this.meterRegistry = meterRegistry;
        this.scheduledTasks = new ConcurrentHashMap();
    }
    
    // 提交即时任务
    public <T> Mono<T> submit(Task<T> task) {
        Counter counter = meterRegistry.counter("task.submitted", 
            "priority", task.getPriority().name());
        counter.increment();
        
        return task.execute()
                  .subscribeOn(virtualThreadScheduler)
                  .doOnSuccess(result -> 
                      logTaskCompletion(task, result, null))
                  .doOnError(error -> 
                      logTaskCompletion(task, null, error))
                  .contextWrite(Context.of("taskId", task.getTaskId()));
    }
    
    // 延迟任务调度
    public <T> Disposable schedule(Task<T> task, Duration delay) {
        String taskId = task.getTaskId();
        Disposable disposable = Mono.delay(delay)
                                  .publishOn(virtualThreadScheduler)
                                  .flatMap(tick -> task.execute())
                                  .subscribe();
        
        scheduledTasks.put(taskId, disposable);
        return disposable;
    }
    
    // 批量任务处理
    public <T> Flux<T> processBatch(List<Task<T>> tasks, int concurrency) {
        return Flux.fromIterable(tasks)
                  .parallel(concurrency)
                  .runOn(virtualThreadScheduler)
                  .flatMap(Task::execute)
                  .sequential()
                  .metrics()
                  .name("batch.task.processing")
                  .tag("batch_size", String.valueOf(tasks.size()));
    }
    
    // 优先级队列调度
    public <T> Flux<T> scheduleWithPriority(
            List<Task<T>> tasks, 
            Function<Task<T>, Integer> priorityExtractor) {
        
        return Flux.fromIterable(tasks)
                  .sort(Comparator.comparing(priorityExtractor))
                  .windowUntilChanged(priorityExtractor)
                  .concatMap(window -> window.flatMap(
                      task -> task.execute()
                                 .subscribeOn(virtualThreadScheduler),
                      5  // 每个优先级级别的并发度
                  ));
    }
    
    private void logTaskCompletion(Task task, Object result, Throwable error) {
        Timer.Sample sample = Timer.start(meterRegistry);
        sample.stop(meterRegistry.timer("task.execution.time",
            "priority", task.getPriority().name(),
            "status", error == null ? "success" : "error"));
        
        if (error != null) {
            log.error("Task {} failed", task.getTaskId(), error);
        } else {
            log.debug("Task {} completed successfully", task.getTaskId());
        }
    }
}