java - CompletableFuture 跨微服务(JVM)

Question

第 1 步:我想要一个 CompletableFuture<String> asyncFuture在 microservice A 中开始通过 say supplyAsync 运行异步任务.

第2步:然后通过手动调用asyncFuture.complete(T value)手动完成相同的 future 对象来自不同的微服务 B，它会被一些异步事件触发。

显然微服务A和微服务B有不同的JVM。实际上，微服务 A 和微服务 B 是运行在 kubernetes 中不同 Pod 上的同一微服务的不同实例。

在第 1 步和第 2 步之间， future 的对象将存储在 Redis 中，微服务 B 可以安全地检索它。

---

快速谷歌搜索后，我想我会尝试以下几个解决方案:

1> HazelCast 的分布式执行器服务，我可以在调用时将其作为第二个参数传入

static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

引用: http://docs.hazelcast.org/docs/2.3/manual/html/ch09.html

2>使用 apache ignite 的共享 executorService

引用:https://apacheignite.readme.io/v1.2/docs/executor-service

不确定两者是否可行？另外我想知道以前有没有人处理过这样的事情？如果是这样，如果您能与我分享您的解决方案，我将不胜感激。

Answer 1

关于Apache Ignite，有很多选择如何协作节点(微服务)。其中之一是连续查询 [1]，它允许监听缓存中发生的数据修改。

例如，在服务 A 上，您可以创建 ContinuousQuery 并等待缓存中的值发生变化:

private String waitForValueChanged(IgniteCache<Integer, String> cache, Integer key) throws InterruptedException {
    ContinuousQuery<Integer, String> qry = new ContinuousQuery<>();

    qry.setInitialQuery(new ScanQuery<>((k, v) -> k == key));

    final CountDownLatch waitForValueChanged = new CountDownLatch(1);
    final AtomicReference<String> result = new AtomicReference<>();

    CacheEntryUpdatedListener<Integer, String> listener = new CacheEntryUpdatedListener<Integer, String>() {
        @Override public void onUpdated(Iterable<CacheEntryEvent<? extends Integer, ? extends String>> iterable) throws CacheEntryListenerException {
            for (CacheEntryEvent<? extends Integer, ? extends String> entry: iterable) {
                result.set(entry.getValue());
            }

            waitForValueChanged.countDown();
        }
    };

    qry.setLocalListener(listener);

    try (QueryCursor<Cache.Entry<Integer, String>> cur = cache.query(qry);) {
        waitForValueChanged.await(60000, TimeUnit.MILLISECONDS);
    }

    return result.get();
}

在服务 B 上，您只需将值放入缓存即可“完成 future ”:

private void completeFuture(IgniteCache<Integer, String> cache, Integer key, String value) {
    cache.put(key, value);
}

这是一个示例项目，展示了连续查询的工作原理 [2]。

[1] https://apacheignite.readme.io/docs#section-continuous-queries

[2] https://github.com/gromtech/ignite-continuous-query-example