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

【问题标题】：Single @StreamListener to route different payload types to generic service classes单个 @StreamListener 将不同的有效负载类型路由到通用服务类
【发布时间】：2019-09-10 21:13:13
【问题描述】：

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

我知道我可以为每个payload类型实现多个@StreamListener方法，例如：

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

但是，对于许多不同的有效负载类型，此解决方案不可扩展。在理想情况下，我不希望使用这个公共库的应用程序甚至知道 Kafka 侦听器方法。相反，我更愿意提供一个 SPI，让消费应用程序实现如下所示的接口，框架将调用实现的 Spring bean 并将正确的负载类型传递给它。

public interface MessageHandlingService<T extends BaseClass> {

    void handleMessage(T payload);

}

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

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

【问题讨论】：

我怀疑单个应用程序可能不“了解 Kafka 侦听器方法”，或者至少您正在使用 Spring Kafka Stream 的事实。毕竟，他们需要提供主题和 SerDe 类来反序列化他们的对象。或者您是否打算提供图书馆中所有这些人的清单？那将是 IMO 的糟糕设计。
除了 Artem 在下面所说的之外，请考虑阅读本节 cloud.spring.io/spring-cloud-static/spring-cloud-stream/…。出于一个简单的原因，您要问的是不可能的。消息以byte[] 的形式出现 - 总是。因此，在我们从处理程序方法（例如，StreamListener 或函数）访问类型信息之前，我们无法知道要转换为哪种类型。

标签： java spring-boot spring-kafka spring-cloud-stream

【解决方案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

【讨论】：