Vertx延迟批处理

Question

如何在Vertx中处理延迟作业列表（实际上是数百个HTTP GET请求，到禁止快速请求主机的有限API）？现在，我正在使用此代码，它被阻止，因为Vertx一次启动所有请求。希望在每个请求之间有5秒的延迟来处理每个请求。

public void getInstrumnetDailyInfo(Instrument instrument,
                                   Handler<AsyncResult<OptionInstrument>> handler) {
  webClient
    .get("/Loader")
    .addQueryParam("i", instrument.getId())
    .timeout(30000)
    .send(
      ar -> {
        if (ar.succeeded()) {
          String html = ar.result().bodyAsString();
          Integer thatData = processHTML(html);
            instrument.setThatData(thatData);
            handler.handle(Future.succeededFuture(instrument));
        } else {
          // error
          handler.handle(Future.failedFuture("error " +ar.cause()));
        }
      });
}

public void start(){
  List<Instrument> instruments = loadInstrumentsList();
  instruments.forEach(
    instrument -> {
      webClient.getInstrumnetDailyInfo(instrument,
            async -> {
              if(async.succeeded()){
                instrumentMap.put(instrument.getId(), instrument);
              }else {
                log.warn("getInstrumnetDailyInfo: ", async.cause());
              }
            });
        });
}

Answer 1

您可以使用任何现成的速率限制器功能，并将其调整为异步使用。

Guava的RateLimitor示例：

    // Make permits available at a rate of one every 5 seconds
    private RateLimiter limiter = RateLimiter.create(1 / 5.0);

    // A vert.x future that completes when it obtains a throttle permit
    public Future<Double> throttle() {
        return vertx.executeBlocking(p -> p.complete(limiter.acquire()), true);
    }

然后

   throttle()
       .compose(d -> {
           System.out.printf("Waited %.2f before running job\n", d);
           return runJob(); // runJob returns a Future result
       });

Answer 2

您可以考虑使用计时器来触发事件（而不是在启动时触发所有事件）。

Vertx有两种变体，

>

. setTimer（）在延迟后触发特定事件

<代码>顶点。setTimer（间隔，新处理程序

和
2<代码>。setPeriodic（），每次经过指定的时间段后都会激发。

vertx.setPeriodic(interval, new Handler<Long>() {});

您正在寻找的似乎是设置周期。

您可以从文档中获得更多信息

对于更复杂的Vertx调度用例，您可以查看Chime或其他调度程序或此模块