feat:多机证书导入功能

2025-05-08 16:46:30 +08:00
parent db80b1dd83
commit 55a9cfbdbe
11 changed files with 773 additions and 19 deletions
@@ -0,0 +1,141 @@
+package com.sunyard.task;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.*;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class AsyncTaskExecutor {
+
+    private final ExecutorService taskExecutor;
+    private final ScheduledExecutorService retryScheduler;
+
+    public AsyncTaskExecutor(int taskThreads, int retryThreads) {
+        this.taskExecutor = Executors.newFixedThreadPool(taskThreads);
+        this.retryScheduler = Executors.newScheduledThreadPool(retryThreads);
+    }
+
+    // 默认配置构造函数
+    public AsyncTaskExecutor() {
+        this(4, 2);
+    }
+
+    public List<TaskOutcome<String>> execTasks(List<TaskContext<String>> taskContexts) {
+        List<CompletableFuture<String>> futures =
+                taskContexts.stream().map(this::retryWithDelay).collect(Collectors.toList());
+        List<TaskOutcome<String>> taskOutcomes = new ArrayList<>();
+        taskContexts.forEach(task ->
+                retryWithDelay(task).whenCompleteAsync((result, ex) -> {
+                    if (ex != null) {
+                        // 提取原始异常
+                        Throwable rootCause = ex instanceof CompletionException ? ex.getCause() : ex;
+                        taskOutcomes.add(new TaskOutcome<>(task, rootCause));
+                    } else {
+                        taskOutcomes.add(new TaskOutcome<>(task, result));
+                    }
+                }));
+        return taskOutcomes;
+    }
+
+    public List<TaskOutcome<String>> execBatchTasks(List<TaskContext<String>> taskContexts) {
+        // 转换每个Future为明确的TaskOutcome类型
+        List<CompletableFuture<TaskOutcome<String>>> futures = taskContexts.stream()
+                .map(task ->
+                        retryWithDelay(task)
+                                .<TaskOutcome<String>>handle((result, ex) -> {
+                                    if (ex != null) {
+                                        // 提取原始异常
+                                        Throwable cause = ex instanceof CompletionException ? ex.getCause() : ex;
+                                        return new TaskOutcome<>(task, cause);
+                                    } else {
+                                        return new TaskOutcome<>(task, result);
+                                    }
+                                })
+                )
+                .collect(Collectors.toList());
+
+        // 等待所有任务完成
+        CompletableFuture<Void> allFutures = CompletableFuture.allOf(
+                futures.toArray(new CompletableFuture[0])
+        );
+
+        // 收集处理结果
+        return allFutures.thenApply(v ->
+                futures.stream()
+                        .map(CompletableFuture::join)
+                        .collect(Collectors.toList())
+        ).join();
+    }
+
+
+    /**
+     * 带延迟的重试逻辑
+     */
+    private <T> CompletableFuture<T> retryWithDelay(TaskContext<T> context) {
+        return CompletableFuture.supplyAsync(context.getTask(), taskExecutor)
+                .handleAsync((result, ex) -> {
+                    if (ex == null) {
+                        return CompletableFuture.completedFuture(result);
+                    } else {
+                        return handleRetry(context, ex);
+                    }
+                }, taskExecutor)
+                .thenCompose(f -> f);
+    }
+
+
+
+    /**
+     * 处理重试逻辑
+     */
+    private <T> CompletableFuture<T> handleRetry(TaskContext<T> context, Throwable ex) {
+        if (context.canRetry()) {
+            context.decreaseRetry();
+            System.out.printf("[重试调度] ID: %s | 延迟: %ss | 剩余重试: %d | 错误: %s%n",
+                    context.getTaskId(),
+                    context.getRetryDelay().getSeconds(),
+                    context.getRetriesLeft(),
+                    ex.getCause().getMessage());
+
+            // 创建延迟重试的Future
+            CompletableFuture<T> delayedRetry = new CompletableFuture<>();
+            retryScheduler.schedule(() -> {
+                retryWithDelay(context).whenComplete((retryResult, retryEx) -> {
+                    if (retryEx != null) {
+                        delayedRetry.completeExceptionally(retryEx);
+                    } else {
+                        delayedRetry.complete(retryResult);
+                    }
+                });
+            }, context.getRetryDelay().toMillis(), TimeUnit.MILLISECONDS);
+
+            return delayedRetry;
+        } else {
+            // 重试耗尽，传递原始异常
+            CompletableFuture<T> failed = new CompletableFuture<>();
+            failed.completeExceptionally(ex);
+            return failed;
+        }
+    }
+
+    /**
+     * 关闭线程池
+     */
+    public void shutdown() {
+        taskExecutor.shutdown();
+        retryScheduler.shutdown();
+        try {
+            if (!taskExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+                taskExecutor.shutdownNow();
+            }
+            if (!retryScheduler.awaitTermination(1, TimeUnit.SECONDS)) {
+                retryScheduler.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}