VERTX事件总线缓慢消耗问题

Question

我们有一个非聚类的VERTX应用程序，我们使用事件总线在垂直之间进行内部通信。

• Verticle A从公共汽车上消耗，执行HTTP请求，并通过公共汽车发送回复。

• Verticle B只是要求执行该HTTP请求。

当Verticle B执行“高”请求量时，就会出现问题。然后，消费者开始收到较慢的事件（大概是因为它们在事件总线中排队）。对于8个请求/秒，总线最多需要3-4秒即可消耗事件。当请求/秒提高时，消耗它可能需要超过30秒的时间，因此触发了总线超时。

问题是，垂直A的执行HTTP操作（〜200ms）的确很快，因此我真的不明白为什么请求被卡在公共汽车上。

我们已经尝试了许多解决方案，但是没有其他方法可以使用：

• 部署垂直a的多个实例，因为工人
• 使用vertx.executeblocking（）执行HTTP请求，

唯一有效的是评论HTTP请求并通过总线返回模拟对象。但是，同样，HTTP请求不得超过200ms，因此它不应阻止总线。

其他信息：我们使用使用Raturofit + OkhttpClient的自动化的REST客户端。由于公司策略，我们无法使用VERTX WebClient，因此我没有尝试此解决方案。

示例

这是我们代码的真正简化版本，因此您可以检查我是否缺少某些内容。

垂直A

// Instantiated in Verticle A
public class EmailSender {

    private final Vertx vertx;
    private final EmailApiClient emailApiClient;

    public EmailSender(Vertx vertx) {
        this.vertx = vertx;
        emailApiClient = ClientFactory.createEmailApiClient();
    }

    public void start() {
        vertx.eventBus().consumer("sendEmail", this::sendEmail);
    }

    public void sendEmail(Message<EmailRequest> message) {
        EmailRequest emailRequest = message.body();
        emailApiClient.sendEmail(emailRequest).subscribe(
            response -> {
                if (response.code() == 200) {
                    EmailResponse emailResponse = response.body();
                    message.reply(emailResponse);

                } else {
                    message.fail(500, "Error sending email");
                }
            });
    }
}

verticer b

// Instantiated in Verticle B
public class EmailCommunications {

    private final Vertx vertx;

    public EmailCommunications(Vertx vertx) {
        this.vertx = vertx;
    }

    public Single<EmailResponse> sendEmail(EmailRequest emailRequest) {
        SingleSubject<EmailResponse> emailSent = SingleSubject.create();
        vertx.eventBus().request(
            "sendEmail",
            emailRequest,
            busResult -> {
                if (busResult.succeded()) {
                    emailSent.onSuccess(busResult.result().body())
                } else {
                    emailSent.onError(busResult.cause())
                }
            }
        );
        return emailSent;
    }
}

Answer 1

我们修复了更改OKHTTPCLIENT配置的问题，因此HTTP请求不会卡住

default void configureOkHttpClient(OkHttpClient.Builder okHttpClientBuilder) {
        ConnectionPool connectionPool = new ConnectionPool(40, 5, TimeUnit.MINUTES);
        Dispatcher dispatcher = new Dispatcher();
        dispatcher.setMaxRequestsPerHost(200);
        dispatcher.setMaxRequests(200);

        okHttpClientBuilder
            .readTimeout(60, TimeUnit.SECONDS)
            .retryOnConnectionFailure(true)
            .connectionPool(connectionPool)
            .dispatcher(dispatcher);
    }