如何使Vertx MongoClient操作同步但不阻塞Java中的事件循环？

Question

我正在尝试使用Vertx MongoClient将新文档保存到MongoDB，如下所示：

MongoDBConnection.mongoClient.save("booking", query, res -> {
    if(res.succeeded()) {
        documentID = res.result();
        System.out.println("MongoDB inserted successfully. + document ID is : " + documentID);
    }
    else {
        System.out.println("MongoDB insertion failed.");
    }
});

if(documentID != null) {

    // MongoDB document insertion successful. Reply with a booking ID
    String resMsg = "A confirmed booking has been successfully created with booking id as " + documentID + 
        ". An email has also been triggered to the shared email id " + emailID;

    documentID = null;

    return new JsonObject().put("fulfillmentText", resMsg);
}
else {
    // return intent response
    documentID = null;
    return new JsonObject().put("fulfillmentText", 
        "There is some issues while booking the shipment. Please start afreash.");
}

上面的代码成功地将查询jsonObject写入MongoDB集合booking。但是，包含此代码的函数始终返回There is some issues while booking the shipment. Please start afreash。

发生这种情况可能是因为MongoClient save()处理程序“res”是异步的。但是，我想基于成功的save()操作和失败的保存操作返回条件响应。

如何在Vertx Java中实现它？

Answer 1

您的假设是正确的，您不要等待数据库的异步响应。你可以做的是将它包装在Future中，如下所示：

  public Future<JsonObject> save() {
    Future<JsonObject> future = Future.future();
    MongoDBConnection.mongoClient.save("booking", query, res -> {
      if(res.succeeded()) {
        documentID = res.result();
        if(documentID != null) {
          System.out.println("MongoDB inserted successfully. + document ID is : " + documentID);
          String resMsg = "A confirmed booking has been successfully created with booking id as " + documentID +
            ". An email has also been triggered to the shared email id " + emailID;
          future.complete(new JsonObject().put("fulfillmentText", resMsg));
        }else{
          future.complete(new JsonObject().put("fulfillmentText",
            "There is some issues while booking the shipment. Please start afreash."))
        }
      } else {
        System.out.println("MongoDB insertion failed.");
        future.fail(res.cause());
      }
    });
    return future;
  }

然后我假设你有最终调用它的端点，例如：

router.route("/book").handler(this::addBooking);

...然后您可以调用save方法并根据结果提供不同的响应

public void addBooking(RoutingContext ctx){
    save().setHandler(h -> {
        if(h.succeeded()){
            ctx.response().end(h.result());
        }else{
            ctx.response().setStatusCode(500).end(h.cause());
        }
    })
}

Answer 2

你可以使用RxJava 2和reactive Mongo Client（io.vertx.reactivex.ext.mongo.MongoClient）

这是一段代码：

部署

public class Deployer extends AbstractVerticle {

   private static final Logger logger = getLogger(Deployer.class);

   @Override
   public void start(Future<Void> startFuture) {
      DeploymentOptions options = new DeploymentOptions().setConfig(config());

      JsonObject mongoConfig = new JsonObject()
            .put("connection_string",
                  String.format("mongodb://%s:%s@%s:%d/%s",
                        config().getString("mongodb.username"),
                        config().getString("mongodb.password"),
                        config().getString("mongodb.host"),
                        config().getInteger("mongodb.port"),
                        config().getString("mongodb.database.name")));

      MongoClient client = MongoClient.createShared(vertx, mongoConfig);

      RxHelper.deployVerticle(vertx, new BookingsStorage(client), options)
            .subscribe(e -> {
               logger.info("Successfully Deployed");
               startFuture.complete();
            }, error -> {
               logger.error("Failed to Deployed", error);
               startFuture.fail(error);
            });
   }
}

BookingsStorage

public class BookingsStorage extends AbstractVerticle {

   private MongoClient mongoClient;

   public BookingsStorage(MongoClient mongoClient) {
      this.mongoClient = mongoClient;
   }

   @Override
   public void start() {
      var eventBus = vertx.eventBus();
      eventBus.consumer("GET_ALL_BOOKINGS_ADDRESS", this::getAllBookings);
   }

   private void getAllBookings(Message msg) {
      mongoClient.rxFindWithOptions("GET_ALL_BOOKINGS_COLLECTION", new JsonObject(), sortByDate())
            .subscribe(bookings -> {
                     // do something with bookings
                     msg.reply(bookings);
                  },
                  error -> {
                     fail(msg, error);
                  }
            );
   }

   private void fail(Message msg, Throwable error) {
      msg.fail(500, "An unexpected error occurred: " + error.getMessage());
   }

   private FindOptions sortByDate() {
      return new FindOptions().setSort(new JsonObject().put("date", 1));
   }
}

HttpRouterVerticle

// inside a router handler: 

 vertx.eventBus().rxSend("GET_ALL_BOOKINGS_ADDRESS", new JsonObject())
                 .subscribe(bookings -> {
                     // do something with bookings
                  },
                    e -> {
                      // handle error
                 });