|
此版本仍在开发中，尚不被认为是稳定的。对于最新的稳定版本，请使用 Spring for Apache Kafka 3.3.9！ |
在 Spring 中管理的生产者拦截器
从 3.0.0 版本开始，当涉及到生产者拦截器时，您可以让 Spring 直接将其作为 bean 进行管理，而不是向 Apache Kafka 生产者配置提供拦截器的类名。
如果您采用这种方法，则需要将此生产者拦截器设置为KafkaTemplate.
下面是一个使用相同MyProducerInterceptor从上面开始，但更改为不使用内部配置属性。
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);
}在发送记录之前，onSend调用生产者拦截器的方法。
一旦服务器发送了发布数据的确认，那么onAcknowledgement方法被调用。
这onAcknowledgement在生产者调用任何用户回调之前调用。
如果您通过 Spring 管理多个此类生产者拦截器，则需要应用于KafkaTemplate，您需要使用CompositeProducerInterceptor相反。CompositeProducerInterceptor允许按顺序添加单个生产者拦截器。
底层ProducerInterceptor实现按添加到CompositeProducerInterceptor.