CompletableFuture和AtomicReference解决异步和同步调用获取结果的问题

背景：存在一个耗时的任务，优先使用异步方式进行调用，但是有时有需要同步调用等待获取任务执行结果。

这里存在下面几个问题：

• 同一任务已经异步提交执行了，现在需要同步执行等待结果
• 同一任务正在同步执行等待结果，后面后进行了异步重复提交，需要确保不会重复调用该任务
• 系统资源有效，耗时任务不能无限的添加执行

解决方案：

1. 可以使用CompletableFuture和AtomicReference来保存异步调用的结果，并在同步调用时检查是否有正在进行的任务。如果有，等待结果；如果没有，启动任务。
2. 使用Semaphore来控制同时能有多少个任务可执行

示例代码：

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicReference;

public class TaskManager {
    private final ConcurrentHashMap<String, AtomicReference<CompletableFuture<String>>> tasks = new ConcurrentHashMap<>();
    private final Semaphore semaphore = new Semaphore(5);  // 限制最多同时执行5个任务

    // 同步执行任务
    public String executeTaskSynchronously(String taskId) {
        AtomicReference<CompletableFuture<String>> taskRef = tasks.computeIfAbsent(taskId, k -> new AtomicReference<>());
        CompletableFuture<String> task = taskRef.get();

        if (task == null) {
            task = startTask(taskId);
            if (!taskRef.compareAndSet(null, task)) {
                task = taskRef.get();
            }
        }

        return task.join();  // 等待异步任务完成并返回结果
    }

    // 异步执行任务
    public CompletableFuture<String> executeTaskAsynchronously(String taskId) {
        AtomicReference<CompletableFuture<String>> taskRef = tasks.computeIfAbsent(taskId, k -> new AtomicReference<>());
        CompletableFuture<String> task = taskRef.get();

        if (task == null) {
            task = startTask(taskId);
            if (!taskRef.compareAndSet(null, task)) {
                task = taskRef.get();
            }
        }

        return task;
    }

    // 实际启动任务
    private CompletableFuture<String> startTask(String taskId) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                semaphore.acquire();  // 获取许可，最多只能有5个任务同时执行
                // 模拟耗时任务
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                return "Result for task " + taskId;
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return "Task interrupted";
            } finally {
                semaphore.release();  // 释放许可，允许下一个任务执行
                tasks.remove(taskId);
            }
        });
    }
}

代码说明：

1. ConcurrentHashMap: tasks 是一个 ConcurrentHashMap，其中的每个任务都有一个唯一的 taskId 作为键。每个任务的执行状态用 AtomicReference<CompletableFuture<String>> 表示。
2. Semaphore: Semaphore 初始化为 5，表示最多允许 5 个任务同时执行。如果一个任务完成了，会释放许可，允许下一个任务执行。
3. 同步调用 (executeTaskSynchronously): 同步调用会获取任务的 CompletableFuture，如果任务还未启动，则启动任务并存储在 AtomicReference 中，然后等待任务完成并返回结果。
4. 异步调用 (executeTaskAsynchronously): 异步调用与同步调用类似，只是不等待任务完成，而是直接返回 CompletableFuture，以便调用者在未来某个时间点处理结果。
5. 任务完成后清理: 当任务完成后，tasks 中与该任务相关的记录会被移除，以便下次可以重新启动任务。