同步并行API请求以从长时间运行的查询返回相同结果而无需多次执行的最佳实践是什么？

Question

我正在做一个由多个客户机调用的后端Spring Boot项目。其中一个功能是合并来自两个不同数据库的数据并返回结果，这可能需要2分钟。

我希望能够对此endpoint进行并发调用，等待已经运行的请求，并返回相同的结果，而无需再次运行查询。

如下所示，我尝试在服务singleton bean中设置一个CompletableFuture字段（我知道这是一种代码味道，因为singleton服务bean应该是无状态的）。

//RestController
    @Async
    @GetMapping
    public CompletableFuture<List<Foo>> getSyncedFoo() {
        return service.syncFoo();
    }

//ServiceImpl

    private CompletableFuture<List<Foo>> syncTask;

    @Override
    @Async
    @Transactional
    public CompletableFuture<List<Foo>> syncFoo() {
        if (this.syncTask == null || this.syncTask.isDone()) {
            this.syncTask = CompletableFuture.supplyAsync(() -> {
                // long running task
                return new ArrayList<>();
            });
        }
        return this.dbaseSyncTask;
    }

我怀疑这是由于完全未来的共享使用，但我不知道我应该采取什么方法。RxJava对可观察的策略有什么用处吗？

Answer 1

我现在找到了一个实用的答案。

@Service
public class FooServiceImpl implements FooService {

    private CompletableFuture<List<Foo>> syncFuture;
    private Observable<List<Foo>> syncObservable;

    @Override
    public Single<List<Foo>> syncFoo() {
        if (syncFuture == null || syncFuture .isDone()) {
            syncFuture = syncFooAsync();
            syncObservable = Observable.fromFuture(syncFuture).share();
        }
        return Single.fromObservable(syncObservable);
    }

    private CompletableFuture<List<Foo>> syncFooAsync() {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(10_000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                return new ArrayList<>();
            }
        });
    }

}

通过使用RxJava库，可以使用observable：：Share方法将创建的observable多播给多个侦听器，并且@RESTController将很乐意处理返回的单个侦听器。

遗憾的是，它仍然在单例中使用state，而单例是由多个线程并发访问的，所以我担心并发问题的情况，比如在创建新订阅的过程中，新请求仍在可观察的完成。