问题标签 [smallrye-reactive-messaging]

我按照官方 Quarkus 消息传递指南创建了一个简单的示例来体验使用 AMQP（Apache Artemis）的反应式消息传递特性。

完整的代码在这里。

该示例正在运行，但存在一个小问题，我必须先启动一个curl来使用消息，然后再使用另一个curl来发送消息。

如果取消消费者，然后发送消息，则会抛出异常。

即使没有连接消费者，也可以在 Apache Artemis 中缓存消息。

我有一个使用 Smallrye Mutiny 的微配置文件应用程序，我需要在 JBoss EAP 7.3 上部署它。我已经应用了 JBoss EAP XP 补丁来启用微配置文件功能。我还在 EAP 中添加了 mutiny jar 作为模块。以下是模块文件内容

反应流的 module.xml 是

但是，在我的应用程序中，我收到以下异常 Failed to define class io.smallrye.mutiny.Multi in Module "deployment.SampleReactive.war" from Service Module Loader: java.lang.NoClassDefFoundError: Failed to link io/smallrye/mutiny /Multi（来自服务模块加载器的模块“deployment.SampleReactive.war”）：org/reactivestreams/Publisher

如何配置 JBoss EAP 以使用使用 Microprofile 构建的反应式应用程序？任何帮助表示赞赏。

编辑：我已经设法让应用程序在 EAP 7.3 上运行。但现在我面临一个不同的问题。应用程序有 REST API，它发出 Uni 和 Multi。REST API 生成 Json。但是当我在 EAP 中运行这个程序时，我看到并反对作为响应而不是 JSON。似乎 EAP 无法将 Uni 和 Multi 转换为 JSON。

我需要添加任何库才能使其正常工作吗？

@Incoming("queue")是否可以使用带有 quarkus 和 smallrye-reactive-messaging注释的相同方法并行处理多个 amqp -消息？

更准确地说，我有以下课程：

使用 application.properties 中的配置：

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

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

我正在开发 Quarkus 应用程序并打算使用 Kafka 接收消息，但是如果应用程序在重试一定次数后无法访问 Kafka 代理，我想停止该应用程序。默认配置是无限次尝试重新连接。在Smallrye Reactive Messaging Kafka的文档中，它说我们可以使用kafka.retry-attempts或mp.messaging.incoming.[channel-name].retry-attempts配置重试次数。我已经尝试了这两种方法，但该应用程序仍在重新启动。

有没有人遇到过类似的问题，或者有人可以帮助我解决问题吗？

我想批处理。在我的用例中，发送 kafka 生产者消息是一一发送的。我想将它们作为消费者应用程序中的列表阅读。我可以在 Spring Kafka 图书馆做到这一点。Spring Kafka 批处理侦听器

有没有办法用quarkus-smallrye-reactive-messaging-kafka库做到这一点？

我尝试了下面的示例，但出现错误。

应用程序属性：

TestConsumerDeserializer：

我的消费者：

在我的场景中，应用程序使用 CDI 生产者从 JMS 队列中读取消息。

我正在使用 quarkus-smallrye 实现。

连接器看起来很简单：

配置：

我尝试了不同的方法来处理消息，但在每种情况下，我都只有一个工作线程。

日志每次显示相同的结果：

我使用了这个文档并搜索了很多 ;-)

我的错误是什么？为什么我只运行一个线程？如果您需要更多信息，请告诉我。

我正在按照以下指南https://smallrye.io/smallrye-reactive-messaging/smallrye-reactive-messaging/2/testing/testing.html在没有 kafka 代理的情况下进行测试。我已经设置了以下 QuarkusTestResource：

并以这种方式使用

但是，我总是收到此错误：

我真的不明白我所缺少的。我很确定这很愚蠢。

我想使用这个扩展：[Quarkus Smallrye Reactive Messaging Kafka]

但是在我的应用程序中，主题的名称是事先不知道的，它是根据运行时从用户那里收到的消息指定的。如何在没有注释的情况下以编程方式指定与主题相关的主题名称和设置？（仅用于向 Kafka 发送消息 -> 生产）

或者这些配置应该在运行时以编程方式设置

因为不认识路，所以用的是原生的Kafka驱动

我想在向反应式消息连接器发送消息的事务性 REST 端点之间传播 JTA 状态（= 事务）。

和

据我了解，上下文传播负责使事务可用（请参阅 参考资料io.smallrye.context.jta.context.propagation.JtaContextProvider#currentContext）。问题似乎是，它currentContext是在订阅时创建的，这在注入点 ( Emitter<String> emitter) 获取其实例时发生。要正确捕获交易还为时过早。

我错过了什么？

顺便说一句，我在使用@Incoming/@Outgoing而不是发射器时遇到了同样的问题。我决定给你这个例子是因为它很容易理解和重现。

我正在使用 Quarkus 将请求代理到另一个 REST 服务。下游服务运行时一切正常。当我关闭下游并通过 Quarkus 发送请求（例如 curl 请求）时，第一个请求被卡住，对于所有其他进一步的 curl 请求，我收到以下错误，即使下游服务恢复，系统也永远不会恢复。

Request failed: java.lang.IllegalStateException: SRMSG00028: The subscription to events has been cancelled

我使用 org.eclipse.microprofile.reactive.messaging.Emitter 和 org.eclipse.microprofile.reactive.messaging.Channel。

我该如何处理这种情况？我希望当下游服务关闭时 curl 请求会超时，并且一旦服务重新启动，我应该会收到 200 个 Ok 响应，而不是“IIlegalStateException 他对事件的订阅已被取消”。

可能是我遗漏了一些简单的东西，但我在 Quarkus 文档中找不到与此相关的任何内容。请指教。

谢谢，