根据gary的评论，我可以设计一个定制的反序列化程序，它可以根据标题中的内容在通用记录和特定记录之间切换。这是它的代码 A .

public class GenericSpecificFlexibleDeserializer implements ExtendedDeserializer<Object> {
  private static final String SPECIFIC_AVRO_READER = "specific.avro.reader";
  private Map<String, ?> config;
  private KafkaAvroDeserializer specificKafkaAvroDeserializer;
  private KafkaAvroDeserializer genericKafkaAvroDeserializer;

  @Override
  public Object deserialize(String topic, Headers headers, byte[] data) {
    List<String> entityTypeHeader = new LinkedList<>();
    if (headers != null) {
      if (new String(headers.lastHeader("message_type").value()).equals("A")){
        return specificKafkaAvroDeserializer.deserialize(topic, data);
      }
    }
      return genericKafkaAvroDeserializer.deserialize(topic, data);
  }

  @Override
  public void configure(Map<String, ?> configs, boolean isKey) {
    this.config = configs;
    HashMap<String, Object> specificConfig = new HashMap<>(config);
    specificConfig.put(SPECIFIC_AVRO_READER, true);

    HashMap<String, Object> genericConfig = new HashMap<>(config);
    genericConfig.put(SPECIFIC_AVRO_READER, false);

    specificKafkaAvroDeserializer = new KafkaAvroDeserializer();
    specificKafkaAvroDeserializer.configure(specificConfig, isKey);
    genericKafkaAvroDeserializer = new KafkaAvroDeserializer();
    genericKafkaAvroDeserializer.configure(genericConfig, isKey);
  }

  @Override
  public Object deserialize(String topic, byte[] data) {
    return deserialize(topic, null, data);
  }

  @Override
  public void close() {
    genericKafkaAvroDeserializer.close();
    specificKafkaAvroDeserializer.close();
  }
}

以及 spring.kafka.cloud.stream.kafka.binder.configuration.value.deserializer 在你的应用程序配置必须设置为 GenericSpecificFlexibleDeserializer 当然。