负载上基于 Spring Cloud Stream 内容的路由

Question

我们正在使用带有 Kafka 和 Avro（本机编码器/解码器）的 Spring Cloud Stream v2.2。我们正在尝试使用基于负载条件的基于内容的路由。据我了解，根据Spring Cloud Stream docs，基于内容的路由只能在标头上实现，因为有效负载在到达条件时还没有经过类型转换过程。因此，除非条件基于字节格式，否则它不会按预期工作。但是，我知道当在本机模式下使用 Avro 时，会跳过消息头并且不处理类型协商。所以我不确定基于内容的路由是否可以按预期在有效负载上工作。

@StreamListener(target = Channels.INPUT, condition =
      "payload.context['type']=='one' or"
          + " payload.context['type']=='two'")
  public void doStuff(TypeOneAndTwoData inputData){
...
channels.outputChannel().send(MessageBuilder.withPayload(inputData).build());
}


@StreamListener(target = Channels.INPUT, condition =
      "payload.context['type']=='three' or"
          + " payload.context['type']=='four'")
  public void doOtherStuff(TypeThreeAndFourData inputData){
...
channels.outputChannel().send(MessageBuilder.withPayload(inputData).build());
}

根据我已经到位的日志记录，我可以看到偶尔会触发doStuff，有时会触发doOtherStuff。但是，似乎它们中的大多数都没有被触发并且消息被跳过。根据输入数据，我确信context.type 只能有“一”、“二”、“三”和“四”这 4 个值，因此根据输入，不可能期望有其他值，但经常我可以在日志中看到以下条目：

Cannot find a @StreamListener matching for message with id: null

我有几个问题：

• 在消息被反序列化为原生 Avro 格式的相应 POJO 类后，负载条件是否对消息起作用？
• 为什么有时条件有效，有时无效？ id: null 有什么意义吗？
• 从线程的角度来看，基于内容的路由如何工作？当我们有两个具有不同条件的 StreamListner 或者它们单线程工作时，是否会运行多个线程？在这种情况下，如何至少一次保证消息传递得到管理？条件是否应该相互排斥？

Answer 1

• 是的；它（当前）受支持。

• id：null 很不幸；对于 Kafka，message.headers['id'] 默认为空 - 无论如何它没有任何意义，所以这条日志消息没有多大用处。

• 问题是没有一个条件与转换后的有效负载匹配

您可以在DispatchingStreamListenerMessageHandler.handleRequestMessage() 中设置断点以找出问题所在。

编辑

我没有回答你的第三个问题。

调用是单线程的；不，一条消息可以匹配多个条件；只有在没有条件匹配时才会得到该日志。请参阅我上面引用的方法。

如果多个条件匹配监听器抛出的任何异常，将停止处理（并调用重试/DLQ 处理等）。