维特。x和Rx Java为服务调用实现缓存

Question

例如，在一个Veritcle中使用UserService，通信通过事件总线进行。来代表它：

class SomeOtherService {

    final UserService userService = new UserService();

    // Mutable state
    final Map<String, Single<String>> cache = new HashMap(); // Not Synchronized ?

    public Single<String> getUserSessionInfo(String id) {
        // Seems it is not save ! :
        return cache.computeIfAbsent(id, _id -> {
                log("could not find " + id + " in cache. Connecting to userService...");
                return userService.getUserSessionInfo(id); // uses generated proxy to send msg to the event bus to call it
            }
        );
    }
}

//在另一台机器上的另一个顶点/微服务的某个地方。

class UserService {

    public Single<String> getUserSessionInfo(String id) {
        return Single.fromCallable( () -> {

            waitForOneSecond();

            log("getUserSessionInfo for " + id);

            if (id.equals("1"))
                return "one";
            if (id.equals("2"))
                return "two";

            else throw new Exception("could not"); // is it legal?

           }
        );
    }

还有客户端代码，我们在其中订阅并决定调度程序：

 final Observable<String> obs2 = Observable.from(new String[] {"1", "1"});


        // Emulating sequential call of 'getUserSessionInfo' to fork in separate scheduler A
        obs.flatMap(id -> {
                    log("flatMap"); // on main thread
                    return someOtherService.getUserSessionInfo(id)
                                           .subscribeOn(schedulerA) // Forking. will thread starvation happen? (since we have only 10 threads in the pool)
                                           .toObservable(); 
                }
        ).subscribe(
                x -> log("next: " + x)
        );

问题是，对于缓存使用HashMap的解决方案有多好（因为这里是共享状态）通过使用computeIfAbent方法？

即使我们使用的是事件循环

我应该使用ReplySubject来实现缓存吗？vert的最佳实践是什么。x rx java？

看起来就像缓存一样。computeIfAbsent是否在EventLoop上运行？是否安全，因为它是连续的？

很抱歉很多问题，我想我可以归结为：在Vert中实现“现金支付”服务呼叫的最佳实践是什么。x和Rx Java？

整个例子如下：

Answer 1

我想我在这里找到了答案：http://blog.danlew.net/2015/06/22/loading-data-from-multiple-sources-with-rxjava/ -

Observable<Data> source = Observable .concat(memory, diskWithCache, networkWithSave) .first(); 

当我用地图保存的时候。放（…）显式而不是使用ComputeFabSent

作为日志，我在事件循环中，我可以安全地使用非同步现金地图