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());
    }
});

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

我找到了解决方法：“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 Verticle 使用来自不同线程池的 JVM 线程，因此不会阻塞事件循环。

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

编辑

请参阅这个简短示例，了解如何启动一个非常简单的工人 Verticle。

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

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());
        });
    }
}