spring kafka：同一主题的不同json有效负载答案

【问题标题】：spring kafka : different json payload on the same topicspring kafka：同一主题的不同json有效负载
【发布时间】：2018-04-05 12:52:49
【问题描述】：

我需要在不使用父类的情况下在同一个 Kafka 主题（例如，Foo、Bar、Car ...）上发送不同的 JSON 有效负载

基于spring kafka文档，我可以在类级别使用@KafkaListener，在方法级别指定@KafkaHandler（doc）

@KafkaListener(topics = "myTopic")
static class MultiListenerBean {

    @KafkaHandler
    public void listen(Foo foo) {
        ...
    }

    @KafkaHandler
    public void listen(Bar bar) {
        ...
    }

    @KafkaHandler
    public void listen(Car car) {
        ...
    }
}

但我得到了这个例外：

[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] ERROR o.s.k.listener.LoggingErrorHandler - Error while processing: ConsumerRecord(topic = myTopic, partition = 1, offset = 0, CreateTime = 1508859519287, checksum = 3297149058, serialized key size = -1, serialized value size = 13, key = null, value = {"foo":"foo"})
org.springframework.kafka.KafkaException: No method found for class java.util.LinkedHashMap
    at org.springframework.kafka.listener.adapter.DelegatingInvocableHandler.getHandlerForPayload(DelegatingInvocableHandler.java:92)
    at org.springframework.kafka.listener.adapter.DelegatingInvocableHandler.getMethodNameFor(DelegatingInvocableHandler.java:147)
    at org.springframework.kafka.listener.adapter.HandlerAdapter.getMethodAsString(HandlerAdapter.java:60)
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:188)
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:72)
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:47)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:794)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:738)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:570)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.lang.Thread.run(Thread.java:745)

我也尝试过这种方法，但它不起作用：

   @KafkaListener(topics = "myTopic")
   public void receive(ConsumerRecord<String, ?> payload)
   {
      if (payload.value() instanceof Foo)
      {
         //
      }
      else if (payload.value() instanceof Car)
      {
         //
      }
   }

我如何配置我的生产者和消费者以使用 spring kafka 在同一个主题上发送不同的 JSON Payload？

【问题讨论】：

请显示，你如何从 JSON 反序列化。

标签： java spring apache-kafka spring-integration spring-kafka

【解决方案1】：

使用 JSON 和多方法侦听器时有一个 catch-22；我们需要知道路由到正确方法的类型；我们无法从方法签名中推断出要转换的类型，因为我们还不知道要调用哪个方法。

您需要一个可以从数据或使用Header（使用 Kafka 11）来确定类型的消息转换器。

然后，在转换器转换为正确的类型之后，我们就可以确定调用哪个方法了。

第二次尝试时，ConsumerRecord 将包含原始 JSON；您需要使用ObjectMapper 进行转换，但同样，您需要提示您要转换为哪种类型（或使用蛮力进行转换，直到成功）。

【讨论】：