在 Spring 中管理的 Producer Interceptor

从版本 3.0.0 开始，当涉及到 producer 拦截器时，你可以让 Spring 直接将其作为 bean 进行管理，而不是将拦截器的类名提供给 Apache Kafka 生产者配置。如果使用此方法，则需要在上设置此生产者侦听器。下面是一个使用上述相同内容的示例，但更改为不使用 internal config 属性。KafkaTemplateMyProducerInterceptorspring-doc.cn

public class MyProducerInterceptor implements ProducerInterceptor<String, String> {

    private final SomeBean bean;

    public MyProducerInterceptor(SomeBean bean) {
        this.bean = bean;
    }

    @Override
    public void configure(Map<String, ?> configs) {
    }

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        this.bean.someMethod("producer interceptor");
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    }

    @Override
    public void close() {
    }

}

@Bean
public MyProducerInterceptor myProducerInterceptor(SomeBean someBean) {
  return new MyProducerInterceptor(someBean);
}

@Bean
public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> pf, MyProducerInterceptor myProducerInterceptor) {
   KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(pf);
   kafkaTemplate.setProducerInterceptor(myProducerInterceptor);
}

在发送记录之前，将调用生产者拦截器的方法。一旦服务器发送了发布数据的确认，就会调用该方法。在创建者调用任何用户回调之前调用。onSendonAcknowledgementonAcknowledgementspring-doc.cn

如果您有多个通过 Spring 管理的此类生产者拦截器，则需要改用。允许按顺序添加单个 producer 拦截器。底层实现中的方法将按照它们添加到 .KafkaTemplateCompositeProducerInterceptorCompositeProducerInterceptorProducerInterceptorCompositeProducerInterceptorspring-doc.cn