如何使用vertx3进行数据库顶部的异步缓存查询和更新？

Question

下面是一个例子，说明如何使用异步vertx-mongo-client在数据库之上使用内存缓存。它尽可能的简约化。它是完全可执行的，重要的东西在顶部。

不幸的是，在建立缓存条目时，没有什么可以阻止对同一个密钥的多个数据库访问和缓存插入。使用标准锁定机制是不可能的，因为它会阻止事件循环。

我需要一个异步缓存和数据库调用，这样cache.get()调用在第一个将实际命中db的调用之后，将“返回到事件循环”(我可以这么说吗？)，“等待”缓存条目可用。(真是可怕的一句，我很抱歉。)

我怎样才能做到这一点？我正在考虑学习一些vertx模块的源代码，例如。vertx-mongo-client，来了解它是如何完成的。但如果有人能在这里给出答案，那就太好了。

package q;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.mongo.MongoClient;

import java.util.HashMap;
import java.util.Map;

public class ExampleVerticle extends AbstractVerticle
{
  @Override
  public void start() throws Exception {
    MongoClient mongoClient = MongoClient.createShared(vertx, new JsonObject().put("db_name", "example_db"));
    SomeCache cache = new SomeCache();

    vertx.eventBus().consumer("sesame_street", messageHandler -> {
      Integer lookUpKey = Integer.valueOf(messageHandler.body().toString());
      JsonObject result = cache.get(lookUpKey);

      if(result != null) {
        messageHandler.reply(result);
        System.out.println("Was served from cache");
      } else {
        mongoClient.findOne("example_collection", new JsonObject().put("_id", lookUpKey),
                new JsonObject(), resultHandler -> {
                  if(resultHandler.succeeded()) {
                    messageHandler.reply(resultHandler.result());
                    cache.put(lookUpKey, resultHandler.result());
                    System.out.println("Value stored in cache");
                  } else {
                    messageHandler.fail(0xBADC0DE, resultHandler.cause().toString());
                  }
                });
      }
    });
  }

  static class SomeCache
  {
    Map<Integer, JsonObject> elMapa = new HashMap<>();

    public void put(Integer key, JsonObject value) {
      elMapa.put(key, value);
    }

    public JsonObject get(Integer key) {
      return elMapa.get(key);
    }
  }

  public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();
    vertx.deployVerticle(new ExampleVerticle(), completionHandler -> {
      if(completionHandler.succeeded()) {
        vertx.eventBus().send("sesame_street", 1, replyHandler -> {
          if(replyHandler.succeeded()) {
            //Comment out this println and you'll maybe see 'Value stored in cache' twice in the console output.
            System.out.println("Yes! " + replyHandler.result().body());
            vertx.eventBus().send("sesame_street", 1);
          }
          vertx.close();
        });
      }
    });
  }
}

Answer 1

Vertx有Asynchronous Lock的概念。这不会阻塞，但是在获得锁时将调用带有锁的处理程序。

看一下：

locks http://vertx.io/docs/apidocs/io/vertx/core/shareddata/Lock.html