使用Spring-boot：Consumer的Kafka通过ConsumerRecord使用java.lang.object（any/all对象）

Question

因此，我使用Spring-Boot2.1.6并集成kafka consumer来使用在该主题上发布的任何类型的消息。作为参考，我在https://docs.spring.io/spring-boot/docs/2.1.6.release/reference/htmlsingle/

所以我在POM中有依赖性：

    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>

我正在application.yml中进行配置

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: foo
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring:
          json:
            value:
              default:
                type: java.lang.Object

 @KafkaListener(topics = "videoEnrichedEvents")
    public void consume(@Payload VideoEnrichedEventsvideoEnrichedEvents){
        LOGGER.debug("Consumed message :"+videoEnrichedEvents);
        System.out.println("Consumed Message :"videoEnrichedEvents);
    }

Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.util.LinkedHashMap] to [com.calamp.connect.vs.model.VideoEnrichedEvents] for GenericMessage [payload={anyotherjson={groups=null, id=0, driverName=from Kusum's console, deviceIdType=null, assetId=null, operatorId=null, avlEventTime=null, videoLink=null, tripId=null, avlEventUuid=null, deviceId=null, appMessageUuid=null, parentAccountList=null, appmsgEventTime=null, enrichedMessage=null, accountId=null}}, headers={kafka_offset=9, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@18213932, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, kafka_receivedTopic=videoEnrichedEvents, kafka_receivedTimestamp=1590218109430}], failedMessage=GenericMessage [payload={anyotherjson={groups=null, id=0, driverName=from Kusum's console, deviceIdType=null, assetId=null, operatorId=null, avlEventTime=null, videoLink=null, tripId=null, avlEventUuid=null, deviceId=null, appMessageUuid=null, parentAccountList=null, appmsgEventTime=null, enrichedMessage=null, accountId=null}}, headers={kafka_offset=9, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@18213932, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, kafka_receivedTopic=videoEnrichedEvents, kafka_receivedTimestamp=1590218109430}]

 @KafkaListener(topics = "videoEnrichedEvents")
    public void consume(@Payload ConsumerRecord consumerRecord){
        LOGGER.debug("Consumed message!!!Full :"+consumerRecord);
        System.out.println("Consumed Message!!! Actual object :"+((LinkedHashMap)consumerRecord.value()));
    }

Answer 1

处理此问题的最简单方法是使用ByteArrayDeserializer和ByteArrayJsonMessageConverterbean（只需将其添加到应用程序上下文中，引导将其连接进来）。

这样，从JSON的转换被推迟到调用方法之前，以便我们知道目标类型是什么。

参见https://docs.spring.io/spring-kafka/docs/2.5.0.release/reference/html/#messaging-message-conversion