CompletableFutureObserver.java

package vc.thinker.utils;

import com.google.common.collect.Maps;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * 合并未来的观察者
 * 用于维护一批需要合并同步结果的线程关系
 *
 * @author HeTongHao
 * @since 2022/8/10 16:43
 */
public class CompletableFutureObserver<T> {
    private final Map<String, CompletableFuture<T>> completableFutureMap = Maps.newConcurrentMap();

    public CompletableFuture<T> subscribe(String key) {
        CompletableFuture<T> completableFuture = new CompletableFuture<>();
        completableFutureMap.put(key, completableFuture);
        return completableFuture;
    }

    public void onMessage(String key, T msg) {
        CompletableFuture<T> completableFuture = completableFutureMap.get(key);
        if (completableFuture != null) {
            completableFuture.complete(msg);
            completableFutureMap.remove(key);
        }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> completableFuture = new CompletableFuture<>();
        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            completableFuture.complete("666");
        }).start();
        String o = completableFuture.get();
        System.out.println(o);
    }
}