Spring 负载上的云流路由

Question

我想为我的微服务使用 spring 云流来处理来自 kafka 的事件。

我阅读了一个主题，该主题可以容纳多个 JSON 有效负载（我有一个主题，因为它收到的所有消息都来自同一主题）。我有不同的云函数来处理不同的负载。

如何根据负载中的属性将传入事件路由到特定函数？

假设我有 JSON 条消息可以具有以下属性：

{
  "type":"A"
  "content": xyz
}

因此输入消息可以有一个属性 A 或B

假设我想在类型为 A 时调用一些 bean 函数，在类型为 B 时调用另一个 bean 函数

Answer 1

从问题中不清楚您使用的是基于消息通道的Kafka binder还是Kafka Streams binder。上面的评论暗示了对 KStream 的一些引用。假设您使用的是基于消息通道的 Kafka 绑定器，您可以选择使用 Spring Cloud Stream 中的消息路由功能。文档的这一部分解释了基本用法：https://docs.spring.io/spring-cloud-stream/docs/3.2.1/reference/html/spring-cloud-stream.html#_event_routing

您可以提供 routing-expression 这是一个 SpEL 表达式来传递正确的属性值。

如果您想要超出 SpEL 表达式所能表达的高级路由功能，您还可以实现自定义 MessageRoutingCallback。有关详细信息，请参阅此示例应用程序：https://github.com/spring-cloud/spring-cloud-stream-samples/tree/main/routing-samples/message-routing-callback

Spring 负载上的云流路由

Spring cloud stream routing on payload

apache-kafka

spring-cloud-stream

spring-cloud-stream-binder-kafka