批量KafkaListener的消费者接收器

Question

我的消费者配置有kafka批处理侦听器配置和@KafkaListener消耗消息列表。我有一个消费者拦截器，我想为每条记录设置唯一ID，并将其值存储在映射诊断上下文（MDC）中。如果我的kafka侦听器消耗单个消息，则唯一ID是正确的。但是我的kafka侦听器消耗消息列表，因此MDC. get（“id”）只获取最后一个值。如何处理它？我的拦截器；

public class KafkaConsumerInterceptor implements ConsumerInterceptor<String, String>{

    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> consumerRecords) {
        ConsumerRecord<String, String> record = consumerRecords.iterator().next();
        setId(record.headers().headers("id"));
        return consumerRecords;
    }

    private void setId(Iterable<Header> idHeader) {
        String id = UUID.randomUUID().toString();
        if (idHeader.iterator().hasNext()) {
            Header header = idHeader.iterator().next();
            id = new String(header.value(), StandardCharsets.UTF_8);
        }
        MDC.put("id", id);
    }
  }

Answer 1

您需要移动MDC。将（）。