问题标签 [smallrye-reactive-messaging]

我正在尝试将我的 Artimis-MQ 客户端迁移到 quarkus 微服务。尝试发送消息时，我始终收到“流未连接”错误。

我试图遵循答案中的建议（使用 microprofile-reactive-messaging）：Quarkus with ActiveMQ？

在我的 build.gradle 中：

示例休息端点，将消息转发到 AMQP

在src/main/resources/application.properties：

我不断地看到“非法状态异常”。我可以从日志中看出 smallrye 找到了 amqp 连接器，但从未真正初始化连接。

目前，我正在尝试在 Kotlin 中编写基于 Maven、Quarkus 和 SmallRye 反应式消息传递的“通知服务”。作为基础，我有一个 Java 示例，它运行良好，我试图将它“翻译”成 Kotlin。

我希望它工作的方式是，我发送一个 HTTP 请求（例如 GET http://localhost:8080/search/ {word}），系统将“单词”（这里是一个字符串）发送到队列“查询” ' 的 Artemis AMQP 消息代理。另一个系统订阅消息代理并根据 HTTP 请求（例如 GET http://localhost:8080/receiver）获取队列“查询”中的“单词”。

然而，在 Kotlin 中，它不起作用，我最好的猜测是，Emitter 不会像在 Java 中那样发送“单词”。

这是我正在使用的代码：

科特林

发送

接收

这是Java版本

发送

接收

这里是 Kotlin 和 Java 的配置文件“application.properties”

一些信息：

• 我根据本指南通过 docker-compose 运行的 AMQP 消息代理。
• Smallrye 反应式消息传递

提前致谢，如果我错过提供信息，请告诉我。

当我需要通过 docker-compose 运行 Kafka 主机和端口时，我无法更改它。我想使用相同的 docker-compose 来运行我的服务和 Kafka。所以我需要更改 Kafka 主机。

我试图提供以下环境变量但没有运气：mp.messaging.outgoing.my-channel.bootstrap.servers="Kafka:9092"

我在 Quarkus/Kafka 指南中使用了相同的 docker-compose： https ://quarkus.io/guides/kafka-guide

Confluent SchemaRegistry 不适用于 Quarkus 和 MicroProfile

最终出现以下错误

我尝试构建一个 Quarkus kafka 消费者应用程序，该应用程序必须反序列化 kafka 主题中可用的 avro 序列化程序消息。当我尝试如下配置模式注册表（Confluent`s）以使用 avro 反序列化 kafka 消息时，我最终得到例外

配置

也尝试过

特性

也尝试过

我不想在 quarkus 中使用 smallrye-kafka 通过 SSL 连接到 kafka 主题。

我的代码在执行时有效mvn compile quarkus:dev。

mvn clean package -Pnative是成功的。但是当我运行本机二进制文件时，它会失败，因为它找不到SaslClientCallbackHandler. 我按照 quarkus 指南使用 SSL 构建本机，并使用 smallrye 扩展配置 kafka。

我的 application.property 文件

我希望本地人也能正常工作。当application.property被评论时，本机运行不会抛出异常。

错误日志：

重现：

1. quarkus-smallrye-reactive-messaging-kafka使用扩展制作新的 quarkus 项目

2. 配置应用程序属性

3. 构建本机并运行它。

注意：我没有任何类src/main/java，当在中找到 SSL 配置时，本机失败application.property

Open Liberty 带有 mpReactiveMessaging-1.0 功能，其中包括一个 Kafka 连接器。

我想与 Azure 服务总线集成，并使用 Smallrye AMQP 连接器（版本 1.0.8）。

当尝试使用@Outgoing-annotation 发送到服务总线队列时，我得到

javax.net.ssl.SSLHandshakeException: Failed to create SSL connection

我在调试 Smallrye 实现时发现了这一点。

我查看了 Proton、vert.x 和 Azure 服务总线 AMQP 文档，但我找不到示例或指示我的一组 MP 配置属性用于工作的 AMQP 服务总线连接有什么问题。

smallrye 文档（来自 https://smallrye.io/smallrye-reactive-messaging/）引用了一个我在编译时遇到问题的示例代码片段......

IE，

——似乎“camel”对象——例如，“camel.subscriber”（上图）——没有与之关联的“subscriber”方法（？）。

编译错误看起来像这样......

我在我的 Maven pom.xml 中包含了以下依赖项（最初只是第一个 - 然后添加了第二个以拼命尝试让这个示例片段正常工作 - 也尝试了 1.0.8 版）

希望熟悉骆驼和/或 smallrye 响应式消息传递的人能够识别导致编译错误的问题吗？

谢谢！

我的骆驼路线应用程序检测到其他应用程序发布的 ampq 消息（发布数字），但无法处理，出现“没有可用的类型转换器”错误。我该如何解决？

路由构建器类

pom.xml

应用程序属性

控制台堆栈跟踪（摘录）

其他注意事项

发布应用程序基于 Quark Ampq 示例：
https ://github.com/quarkusio/quarkus-quickstarts/tree/master/amqp-quickstart/src/main/java/org/acme/quarkus/sample

技术

爪哇 8

夸库斯

小黑麦

骆驼

行家

亲爱的，我正在尝试做一些事件驱动的微服务。目前，当使用 Quarkus 和 Smallrye-Reactive 消息传递扩展接收到消息时，我能够使用来自 Kafka 的消息并更新数据库记录。我想要进一步实现的是能够在成功的情况下向其他主题发送消息，否则向错误主题发送消息。我知道我们可以使用 return 和 @outgoing 注释来发出新消息，但我认为它不适合我的用例。如果在使用消息时发生错误，我需要一个指导。我应该将消息返回到原始主题（通过不确认消息）还是应该使用它并向不同的主题产生错误消息以回滚原始事务。

这是我的代码：

另外，如果您有更好的代码，请提交，我还是反应式编程的新手 :)。

问题：发布者应用程序不断向 Artemis 代理发布 AMQP 消息 - 但是，消费者应用程序在第 127 个之后停止接收消息。

Fwiw - 巧合的是，注意到 Smallrye 网页的摘录引用：

“可以使用使用 MicroProfile Config 设置的 smallrye.messaging.emitter.default-buffer-size 属性来配置默认缓冲区大小 (127)。请注意，此值仅在不使用 OnOverflow 时应用。”

（来自： https : //smallrye.io/smallrye-reactive-messaging/ - 部分标题“14.3.@Channel”）

但是，我尝试使用 microprofile-config.properties 添加属性更改（看看我是否至少可以更改“127”数字），但是没有效果。

即，这些似乎都不起作用......

发布者（服务器）控制台：

订户（服务器）控制台：

尝试在 Artemis 配置文件中操作“信用”属性：broker.xml ......特别是 AMQP 的“接受者” - 但是无济于事：

发布者应用程序：

src/main/resources/META-INF/microprofile-config.properties（发布者）：

订阅者应用程序：

src/main/resources/META-INF/microprofile-config.properties（订阅者）：

pub 和 sub 的 pom.xml 信息（基本上重复 - pub 显示，如下）：

Apache Artemis broker.xml：

使用的技术：

• 爪哇 11
• 阿帕奇骆驼 3.0
• Smallrye 反应式消息传递 API
• Apache Artemis 代理 2.10.1

笔记：

• 我们分别为发布者应用程序和服务器应用程序使用单独的 Helidon 服务器。
• 使用 Qpid 代理也会出现同样的问题。