如果是断断续续的，很可能回复时间太长了。这个信息似乎很清楚
可能超时，或使用共享回复主题
每个客户端示例必须使用自己的回复主题或专用分区。
编辑
如果收到的邮件的相关id与this.futures（待定回复）中当前的条目不匹配，则会获取日志。这只能在以下情况下发生：
请求超时（在这种情况下会有相应的警告日志）。
模板是stop（）ped（在这种情况下，this.futures被清除）。
由于某种原因（不应该发生），已处理的回复将被重新传递。
在将密钥添加到this.futures之前会收到回复（因为它是在发送（）或删除记录之前插入的，所以无法执行）。
服务器端为同一请求发送两个或多个回复。
其他应用程序正在向同一回复主题发送数据。如果您可以用调试日志记录来重现它，这将很有帮助，因为这样我们就可以在send上记录相关键。

在使用者上使用属性@header（kafkaheaders.correlation\u id）修复的问题

@KafkaListener(topics = "${kafka.topic.model}")
@SendTo("replymodeltopic")
@Override
public Model receive(ConsumerRecord<String, model> record, @Header(KafkaHeaders.CORRELATION_ID) byte[] correlation) { 
    record.headers().add(KafkaHeaders.CORRELATION_ID, correlation);
    return record.value();
}

在我的配置上我有

@Configuration
@EnableKafka
public class ReceiverConfig {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${kafka.tunnel.group}")
    private String tunnelGroup;

    @Value("${kafka.topic.json.reply}")
    private String jsonTopicReply;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, tunnelGroup);

        return props;
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return props;
    }

    @Bean
    public ConsumerFactory<String, Model> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(), new JsonDeserializer<>(Model.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Model> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Model> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setReplyTemplate(replyTemplate());
        return factory;
    }

    @Bean
    public ProducerFactory<String, Model> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, Model> replyTemplate() {
        KafkaTemplate<String, Model> kafkaTemplate = new KafkaTemplate<>(producerFactory());
        kafkaTemplate.setDefaultTopic(jsonTopicReply);
        return kafkaTemplate;
    }

}