此版本仍在开发中,尚未被视为稳定版本。对于最新的稳定版本,请使用 Spring for Apache Kafka 3.3.0! |
在 Spring 中管理的 Producer Interceptor
从版本 3.0.0 开始,当涉及到 producer 拦截器时,你可以让 Spring 直接将其作为 bean 进行管理,而不是将拦截器的类名提供给 Apache Kafka 生产者配置。
如果使用此方法,则需要在 上设置此生产者侦听器。
下面是一个使用上述相同内容的示例,但更改为不使用 internal config 属性。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 管理的此类生产者拦截器,则需要改用。 允许按顺序添加单个 producer 拦截器。
底层实现中的方法将按照它们添加到 .KafkaTemplate
CompositeProducerInterceptor
CompositeProducerInterceptor
ProducerInterceptor
CompositeProducerInterceptor