如何使用相同的@In的方法并行处理多个AMQP消息

Question

是否可以使用使用夸克和小黑-反应-消息注释的相同方法并行处理多个amqp-消息？

更准确地说，我有以下类:

@ApplicationScoped
public class Receiver {
    @Incoming("test-queue")
    public void process(String input) {
        System.out.println("start processing:" + input);
        try {
            Thread.sleep(10_000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("end processing:" + input);
    }
}

使用应用程序属性中的配置：

amqp-host: localhost
amqp-port: 5672
amqp-username: quarkus
amqp-password: quarkus
mp.messaging.incoming.test-queue.connector: smallrye-amqp
mp.messaging.incoming.test-queue.address: test-queue

现在，我想通过配置来定义可以并行处理多少条消息。例如，在4核cpu上，它应该并行运行4个内核。

目前，我只能添加4个具有不同名称的方法副本，以允许这种并行性，但这是不可配置的。

Answer 1

我刚刚遇到了相同的场景，下面是规范打算如何处理并发：来自eclipse Microprofile规范

基本上，不是用这样的方法创建一个类:

@Incoming("test-queue")
public void process(String input) {}

你有2个这样的类：

@ApplicationScoped
public class MessageSubscriberProducer {

    @Incoming("test-queue")
    public Subscriber<String> createSubscriber() {
        return new SubscriberImpl();
    }
}

public class SubsciberImpl implements Subscriber<String> {

    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        this.subscription.request(4);  // this tells how many messages to grab right away
    }

    @Override
    public void onNext(String val) {
        // do processing
        this.subscription.request(1);  // grab 1 more
    }
}

这样做还有一个好处，就是可以将处理代码从vert.x事件循环线程转移到工作线程池中。

Answer 2

我不确定，但我不认为反应式消息传递支持您的要求。

但是，您可以以另一种方式做您想做的事。我认为这也是使用消息传递的更好整体模式。

http://small rye . io/small rye-reactive-messaging/small rye-reactive-messaging/2.5/amqp/amqp . html # amqp-inbound

查找带有CompletionStage和显式ack()的示例。这种变体是异步的，所以如果您将它与Java现有的并发设施结合起来，您将获得高效的并行处理。

我会将传入的工作发送给执行器，然后在完成时执行任务ack（）。