ActiveMQ Artemis:从smallrye-reactive-messaging (AMQP)以编程方式创建队列

Question

我的Quarkus微服务正在使用smallrye反应式消息传递库中的AMQP连接器来生成发送到从运行的ActiveMQ Artemis代理的消息

Docker镜像。The The The

反应式消息库

文档提到了使用动态地址名称的可能性。我在REST资源中使用以下(Kotlin)代码：

@Inject
    @Channel("task-finished")
    lateinit var taskFinishedEmitter: MutinyEmitter

    @POST
    @Produces(MediaType.TEXT_PLAIN)
    fun doSomethingAndInform(@RestForm customerId: String): Uni {

       // leaving out the actual messageText computation...

       val messageText: String = "DUMMY MESSAGE"
       val metadata: OutgoingAmqpMetadata = OutgoingAmqpMetadata.builder()
          .withDurable(true)
          .withCorrelationId(customerId)
          .withAddress("anycast://my-custom-address")
          .build()

       val message: Message = Message.of(messageText,
            {
                logger.info("message acked")
                CompletableFuture.completedFuture(null)
            },
            {
                logger.info("message nacked: {}", it.message)
                CompletableFuture.completedFuture(null)
            }
       )

       taskFinishedEmitter.send(message.addMetadata(metadata))
       return Uni.createFrom().item("DONE")
    }

连接器在

amqp-host=localhost
amqp-port=5672
amqp-username=adm
amqp-password=***

mp.messaging.outgoing.task-finished.connector=smallrye-amqp

ActiveMQ Artemis确实创建了

动态寻址，但是它不会创建任何绑定到它的队列，消息最终是

未路由

..。

The The The

配置文件包含在

部分

tcp://0.0.0.0:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpMinLargeMessageSize=102400;amqpDuplicateDetection=true;anycastPrefix=anycast://

   true

我尝试将队列名称与地址一起传递

.withAddress("anycast://my-custom-address::my-queue")

但这并没有什么不同。

以编程方式创建的队列和传递给它的消息缺少什么？另外，当消息丢失(未路由)时，Artemis为什么要确认消息？

更新:从Artemis web界面附加屏幕截图

​

Answer 1

默认情况下，Artemis会将ActiveMQ客户端发送的消息视为

组播

(即发布/订阅)。多播的语义规定，发布到代理的每个消息都将被分派到每个订阅者。然而，由于没有订阅者，消息被简单地丢弃(即未路由)。

因为你在你的地址前面加了前缀

您应该在"amqp“上配置该前缀。

在

使用

参数，例如：

tcp://0.0.0.0:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpMinLargeMessageSize=102400;amqpDuplicateDetection=true;anycastPrefix=anycast://

您还可以更改用于自动创建的资源的默认路由类型，例如：

ANYCAST
   ANYCAST