"Found ambiguous parameter type Void"？ Spring 与 Project Reactor 集成

Question

我正在使用带有 Spring 集成的 Project Reactor 从 Kafka 读取并写入 MongoDB，并且我的 Kafka 使用效果很好，但是 .handle(MongoDb.reactiveOutboundChannelAdapter(mongoFactory)) 卡住了。我看到这个函数的内部代码是new ReactiveMongoDbStoringMessageHandler(mongoFactory))，所以我尝试了以下方法（我有一个transform()方法从ConsumerRecord转换为Mono<String>，使用 @Transformer 注释）：

    public IntegrationFlows writeToMongo() {
         return IntegrationFlows.from(kafkaChannel)
              .transform(this)
              .handle(new ReactiveMongoDbStoringMessageHandler(mongoFactory))
              .get();
    }

代码遵循文档 https://docs.spring.io/spring-integration/reference/html/mongodb.html#mongodb-reactive-channel-adapters。 我得到的错误是： java.lang.IllegalArgumentException: Found ambiguous parameter type [class java.lang.Void] for method match: 然后是一长串函数。这可能发生的任何原因？

Answer 1

如果您不打算订阅返回的 Mono，则不能执行 new ReactiveMongoDbStoringMessageHandler(mongoFactory)。 .handle(MongoDb.reactiveOutboundChannelAdapter(mongoFactory)) 是正确的做法，因为它将 ReactiveMongoDbStoringMessageHandler 包装到 ReactiveMessageHandlerAdapter 中以进行自动订阅。

不过我认为您真正的问题在于 .transform(this)。我相信你在这个 class 中有很多方法，所以方法名称要更具体。而这与 Project Reactor 无关。不确定为什么会在发送到 ReactiveMongoDbStoringMessageHandler 之前尝试转换为 Mono...您可能在提供不是 [=24= 的有效载荷（ConsumerRecord ？）时遇到问题] 用于保存到集合中的映射实体。

Answer 2

我也遇到过。解决方案是（注意 ReactiveMessageHandlerAdapter）：

public IntegrationFlows writeToMongo() {
     return IntegrationFlows.from(kafkaChannel)
          .handle(new ReactiveMessageHandlerAdapter(new ReactiveMongoDbStoringMessageHandler(mongoFactory)))
          .get();
}

当 this issue 将被解析时，可以用 handleReactive() 代替。