Quarkus如何启动多个消息消费者？

Question

当我编写消息消费者（如 Kafka/AMQP 等）时，我正在尝试从 Vert.x 迁移到 Quarkus 和 Vert.x。我必须扩展垂直数量以最大限度地提高跨多个核心的性能，即垂直扩展 - 是这在 Quarkus 中可能吗？我看到一个类似的问题here，但没有得到回答。

例如，对于 Kafka，我可能会在一个垂直领域内创建一个消费者，然后在进行性能测试以确定这是最佳状态后，scale 该垂直领域说 10 次（即指定部署中的实例数为 10）数字。我的理解是，默认情况下，1 个垂直 = 1 个事件循环，并且不会跨多个内核扩展。

我知道在 Quarkus 中使用 Vert.x 垂直是 possible，但是还有其他方法可以扩展诸如跨多个核心的 Kafka 消费者数量之类的东西吗？

我看到这种类型的可扩展性可以为 Quarkus HTTP 之类的东西配置，但我找不到任何关于消息消费者的信息。

Answer 1

这是 Vert.x Verticle 方法，总的来说我很满意，但我希望有更好的文档来说明如何做到这一点。

更新 - 字段注入不适用于此示例，但构造函数注入确实有效。

假设我想注入这个

@ApplicationScoped
public class CoffeeRepositoryService {

    public CoffeeRepositoryService() {
        System.out.println("Injection succeeded!");
    }
}

这是我的 Verticle

package org.acme;

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.vertx.core.AbstractVerticle;
import io.vertx.core.impl.logging.Logger;
import io.vertx.core.impl.logging.LoggerFactory;
import io.vertx.mutiny.core.eventbus.EventBus;
import io.vertx.mutiny.rabbitmq.RabbitMQClient;
import io.vertx.mutiny.rabbitmq.RabbitMQConsumer;
import io.vertx.rabbitmq.QueueOptions;
import io.vertx.rabbitmq.RabbitMQOptions;

public class RQVerticle extends AbstractVerticle {

  private final Logger LOGGER = LoggerFactory.getLogger(org.acme.RQVerticle.class);

  //This doesn't work - returns null
  @Inject
  CoffeeRepositoryService coffeeRepositoryService;

  RQVerticle() {} // dummy constructor needed
   
  @Inject // constructor injection - this does work
  RQVerticle(CoffeeRepositoryService coffeeRepositoryService) {
    //Here coffeeRepositoryService is injected properly
  }

  @Override
  public Uni<Void> asyncStart() {
    LOGGER.info(
        "Creating RabbitMQ Connection after Quarkus successful initialization");

    RabbitMQOptions config = new RabbitMQOptions();
    config.setUri("amqp://localhost:5672");
    RabbitMQClient client = RabbitMQClient.create(vertx, config);
    
    Uni<Void> clientResp = client.start();
    clientResp.subscribe()
        .with(asyncResult -> {
          LOGGER.info("RabbitMQ successfully connected!");
              
        });

      return clientResp;
  }
}

主类 - 注入不能这样工作

package org.acme;

import io.quarkus.runtime.Quarkus;
import io.quarkus.runtime.QuarkusApplication;
import io.quarkus.runtime.annotations.QuarkusMain;
import io.vertx.core.DeploymentOptions;
import io.vertx.mutiny.core.Vertx;

@QuarkusMain
public class Main {

  public static void main(String... args) {
    Quarkus.run(MyApp.class, args);
  }

  public static class MyApp implements QuarkusApplication {

    @Override
    public int run(String... args) throws Exception {
      var vertx = Vertx.vertx();
      System.out.println("Deployment Starting");

      DeploymentOptions options = new DeploymentOptions()
          .setInstances(2);
      vertx.deployVerticleAndAwait(RQVerticle::new, options);

      System.out.println("Deployment completed");

      Quarkus.waitForExit();
      return 0;
    }
  }
}

具有有效注入但不能部署多个实例的主类

package org.acme;

import io.quarkus.runtime.StartupEvent;
import io.vertx.mutiny.core.Vertx;
import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.event.Observes;
import org.jboss.logging.Logger;

@ApplicationScoped
public class MainVerticles {

  private static final Logger LOGGER = Logger.getLogger(MainVerticles.class);

  public void init(@Observes StartupEvent e, Vertx vertx, RQVerticle verticle) {
    public void init(@Observes StartupEvent e, Vertx vertx, RQVerticle verticle) {
    DeploymentOptions options = new DeploymentOptions()
    .setInstances(2);
    vertx.deployVerticle(verticle,options).await().indefinitely();
  }
}

Std Out - 第一个主类看起来不错

2021-09-15 15:48:12,052 INFO [org.acm.RQVerticle] (vert.x-eventloop-thread-2) 在 Quarkus 成功初始化后创建 RabbitMQ 连接

2021-09-15 15:48:12,053 INFO [org.acm.RQVerticle] (vert.x-eventloop-thread-3) 在 Quarkus 成功初始化后创建 RabbitMQ 连接

Std Out - 第二个主要课程

2021-09-22 15:48:11,986 错误 [io.qua.run.Application]（Quarkus 主 线程）无法启动应用程序（使用配置文件开发）： java.lang.IllegalArgumentException：无法指定 > 1 个实例 已经创建的verticle