java - 单个 @StreamListener 将不同的有效负载类型路由到通用服务类

Question

我正在为多个 Spring Boot 应用程序实现一个通用库，它将为 Kafka 提供一个简单的接口(interface)。我选择了 Spring Cloud Stream 来集成 Kafka。该库必须支持多种不同的有效负载类型，但是我很难理解如何以干净且通用的方式完成此操作。

我知道我可以实现多个@StreamListener每种负载类型的方法，例如:

    @StreamListener(target = SinkBindings.INPUT, condition = "headers['X-Target-Type'] matches '.*\\.Person'")
    public void listenForMessage(Person payload) {
        // do stuff with payload
    }

但是，此解决方案无法针对许多不同的有效负载类型进行扩展。在理想的情况下，我不希望使用这个通用库的应用程序知道 Kafka 监听器方法。相反，我更愿意提供一个 SPI，让消费应用程序实现如下所示的接口(interface)，并且框架将调用实现的 Spring bean 并向其传递正确的负载类型。

public interface MessageHandlingService<T extends BaseClass> {

    void handleMessage(T payload);

}

对于每种有效负载类型，我希望避免实现单独的 @StreamListener方法，但是由于 Java 的类型删除，这似乎相当困惑。我想知道是否有一种现实的方法可以使用单个流监听器方法将不同的有效负载对象路由到通用服务 bean？

出现的另一个想法是添加 boolean support(Class<?> clazz)方法到上面的接口(interface)，这将允许库检查哪个类支持有效负载，但这似乎有点“hacky”(？？？)。

Answer 1

传入消息的有效负载将完全转换为预期类型:@StreamListener 无法执行其他操作。使用 @StreamListener 的 Spring Cloud Stream 方法取决于方法参数的预期类型。这是一个转换信号。您无法将有效负载转换为类型，然后尝试仅使用 @StreamListener 来确定要调用的方法。类级别的 @KafkaListener 和方法上的 @KafkaHandler 可以在这里为您提供帮助:https://docs.spring.io/spring-kafka/docs/2.3.0.RC1/reference/html/#class-level-kafkalistener .

完全成熟的 Spring Integration 流程及其转换功能也可用于此类逻辑。

在文档中查看有关内容类型协商的更多信息:https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/2.2.1.RELEASE/spring-cloud-stream.html#content-type-management