我当前的设置具有以下配置：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> myKafkaListenerContainerFactory(
        ConsumerFactory<String, String> consumerFactory) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new
            ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setMessageConverter(stringJsonMessageConverter());
    return factory;
}

其中stringjsonmessageconverter

@Bean
public StringJsonMessageConverter stringJsonMessageConverter() {
    return new StringJsonMessageConverter(objectMapper());
}

使用我的对象Map器

@Bean
public ObjectMapper objectMapper() {
    return new ObjectMapper()
            .registerModule(new JavaTimeModule())
            .registerModule(myCustomJacksonModules())
            .configure(FAIL_ON_UNKNOWN_PROPERTIES, false)
            .configure(ACCEPT_SINGLE_VALUE_AS_ARRAY, true)
            .configure(WRITE_DATES_AS_TIMESTAMPS, false);
}

通过此配置，我可以发布为：

...
headers = new MessageHeaders(singletonMap(TOPIC, topic));
Foo foo = ....
Message<?> message = new GenericMessage<>(foo, headers);
kafkaTemplate.send(message);

消费为：

@KafkaListener(topics = "myTopic",
        groupId = "g1",
        containerFactory = "myKafkaListenerContainerFactory")
public void onMessageReceived(Foo foo) {
    ... works with foo here
}

在本例中，如果foo是具体的类，那么这个方法很好用。然而，如果foo是一个抽象，而作为消息发送的是它的子类

F1 extends Foo; F2 extends Foo ...
... // publisher
headers = new MessageHeaders(singletonMap(TOPIC, topic));
F1 f1 = ....
Message<?> message = new GenericMessage<>(f1, headers);
kafkaTemplate.send(message);
... // Listener
@KafkaListener(topics = "myTopic",
        groupId = "g1",
        containerFactory = "myKafkaListenerContainerFactory")
public void onMessageReceived(Foo foo) {
    ... wont work
}

在我的使用者中将foo声明为type将失败，但这会起作用：

@KafkaListener(topics = "myTopic",
        groupId = "g1",
        containerFactory = "myKafkaListenerContainerFactory")
public void onMessageReceived(Map<String,Object> fooAsMap) {
    ... this works too
}

除了上面提到的使用Map的选项外，我还尝试了以下方法：

- class level kafka listener ([from here)][1]; it still requires method with param of type Map
- custom deserializer registered in object mapper?

有没有办法在我发送消息时指定目标类型？