VertX 多个 Worker 实例处理同一条消息

Question

我有一个简单的 Vertx worker vertical，它有 4 个实例用于缩放，如下定义。当多个请求到来时，我期望每个工作实例将同时处理单个请求（一次 4 个请求）。

Vertx vertx = Vertx.vertx();
DeploymentOptions deploymentOptions = new DeploymentOptions()
                                                     .setWorker(true)
                                                     .setInstances(4);
vertx.deployVerticle(MailVertical.class.getName(), deploymentOptions);

一些代码将收到的邮件信息与发布方法联系起来。

// This is executed once per incoming message    
this.vertx.eventBus().publish("anAddress", messageString);

用于记录传入邮件消息的垂直消费者代码

public class MailVertical extends AbstractVerticle {

private static final Logger logger = LoggerFactory.getLogger(MailVertical.class);

@Override
public void start(Promise<Void> future) {

    logger.info("Welcome to Vertx: MailVertical.");

    vertx.eventBus().consumer("anAddress", message -> {

        String msg = message.body().toString();
        
        for(int i=0;i<50;i++){
            logger.info(msg);
        }
        try {
            updateStatusInDB(msg);
        } catch (SQLException e) {
            e.printStackTrace();
        }
    });
}

但是，我观察到每个请求都由所有 4 个实例同时处理，即如果有 1 个请求进入，则总共发生 4 个处理事件 = 200 条日志消息。

...
...
[vert.x-worker-thread-7] INFO com.vertx.mailproject.MailVertical - {"msg":"sample message for app-2123x mail notification","appname":"app-1","msgid":"64fd684b-45a8-48c7-9526-4606d6adc311"}
[vert.x-worker-thread-6] INFO com.vertx.mailproject.MailVertical - Msg: sample message for app-2123x mail notification updated to SENT in DB.
[vert.x-worker-thread-7] INFO com.vertx.mailproject.MailVertical - Msg: sample message for app-2123x mail notification updated to SENT in DB.
[vert.x-worker-thread-4] INFO com.vertx.mailproject.MailVertical - Msg: sample message for app-2123x mail notification updated to SENT in DB.
[vert.x-worker-thread-5] INFO com.vertx.mailproject.MailVertical - Msg: sample message for app-2123x mail notification updated to SENT in DB.

任何建议我在这里做错了什么？或者这是预期的。

vertx-core = 4.2.3

Answer 1

是的，这确实是一种正确的行为。您正在启动 4 个工作线程，然后将它们全部注册到同一地址“anAddress”。

此处使用发布订阅模式。使用方法 publish() 所有注册的 consumer/handler 都将收到此事件。见 Publish / subscribe messaging

如果您只希望其中一个工作人员接收此事件，则使用Point-to-point and Request-Response messaging. Basically replace publish() with send()

但是看看你的代码，我建议不要使用 worker verticle，而是在标准 verticle

中使用 executeBlocking

VertX 多个 Worker 实例处理同一条消息

VertX multiple Worker Instances processing same message

vert.x

vertx-verticle