java - Vertx 中的异步自定义方法

Question

我正在尝试在 Vert.x 中创建自己的异步自定义方法，类似于他们的代码:

// call the external service
WebClient client = WebClient.create(vertx);

client.get(8080, "localhost:8080", "/fast").send(ar -> {
    if (ar.succeeded()) {

        HttpResponse<Buffer> response = ar.result();
        System.out.println("response.bodyAsString()" + response.bodyAsString());

    } else {
        System.out.println("Something went wrong " + ar.cause().getMessage());
    }
});

当您运行此代码时，线程会 hibernate ，而不会阻塞所有者线程，并且在端点响应时执行提供的处理程序。

我找到了使用“executeBlocking”、“createSharedWorkerExecutor.executeBlocking”并使用总线来实现此目的的方法，但在所有这些方法中线程都被阻塞。

我正在寻找一种不阻塞容器线程的方法，但我没有找到。有一个帖子:

How can I implement custom asynchronous operation in Vert.x?

我尝试这样做，但它也阻塞了线程:

vertx.runOnContext(v -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
    }
    handler.handle(Future.succeededFuture("result"));
});

上面的代码在同一个线程中运行，但不会同时运行，因此我假设线程被阻塞。

有什么办法吗？

Answer 1

调用 Thread.sleep() 的方式将使当前的 JVM 线程进入休眠状态，从而有效地阻止在同一线程中运行的当前 vert.x 事件循环。这不是 vert.x 中执行阻塞代码的惯用方式。

请参阅此处:“The Golden Rule - don't block the event loop”。

如果您必须运行阻塞代码，例如 Thread.sleep()，您应该使用 worker verticle 来实现该代码。 。 Worker verticles 使用来自不同线程池的 JVM 线程，因此不会阻塞事件循环。

您上面发布的第一个代码示例没有使用阻止代码，正如您正确描述的那样。它使用异步、非阻塞事件处理程序的惯用方式。

编辑

查看这个简短的示例，了解如何启动一个非常简单的 worker 垂直体。

WorkerVerticle 类中的代码永远不会阻塞事件循环。您可以通过设置 DeployerVerticle 中所示的正确选项，在 Verticle 部署期间使其成为工作线程。

public class DeployerVerticle extends AbstractVerticle {

    @Override
    public void start() throws Exception {
        System.out.println("Main verticle has started, let's deploy another...");

        // Deploy it as a worker verticle
        vertx.deployVerticle("io.example.WorkerVerticle", 
                             new DeploymentOptions().setWorker(true));
    }
}


// ----

package io.example;
/**
 * An example of a worker verticle
 */
public class WorkerVerticle extends AbstractVerticle {

    @Override
    public void start() throws Exception {
        System.out.println("[Worker] Starting in " +
            Thread.currentThread().getName());

        // consume event bus messages sent to address "sample.data"
        // reply with incoming message transformed to upper case
        vertx.eventBus().<String>consumer("sample.data", message -> {

            Thread.sleep(1000); // will not block the event loop
                                // but only this verticle

            System.out.println("[Worker] Consuming data in " + 
                Thread.currentThread().getName());
            String body = message.body();
            message.reply(body.toUpperCase());
        });
    }
}