使用Reactor堆芯发布服务器在Spring Cloud Stream中回调生产者

Question

我编写了一个Spring-Cloud-Stream应用程序，其中制作人将消息发布到指定的Kafka主题。我的问题是，如何添加生产者回调以接收确认/确认消息已成功发布到主题上？就像我们在《KafkaSpring》中所做的那样。发送（记录，新回调{…}） （维护异步生产者）。下面是我的代码：

private final Sinks.Many<Message<?>> responseProcessor = Sinks.many().multicast().onBackpressureBuffer();

@Bean
public Supplier<Flux<Message<?>>> event() {
        return responseProcessor::asFlux;
    }    

public Message<?> publishEvent(String status) {
  try {
      String key = ...;
      response = MessageBuilder.withPayload(payload)
                  .setHeader(KafkaHeaders.MESSAGE_KEY, key)
                  .build();
      responseProcessor.tryEmitNext(response);
  } 

如何确保tryEmitNext已成功写入主题？

实施ProducerListener是否是一种解决方案，是否可能？在Spring Cloud Stream中找不到具体的解决方案/文档

更新

我现在已经实现了下面的内容，似乎按预期工作

@Component
public class MyProducerListener<K, V> implements ProducerListener<K, V> {

    @Override
    public void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata) {
        // Do nothing on onSuccess
    }

    @Override
    public void onError(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata, Exception exception) {
        log.error("Producer exception occurred while publishing message : {}, exception : {}", producerRecord, exception);
    }

}

 @Bean
    ProducerMessageHandlerCustomizer<KafkaProducerMessageHandler<?, ?>> customizer(MyProducerListener pl) {
        return (handler, destinationName) -> handler.getKafkaTemplate().setProducerListener(pl);
    }

Answer 1

请参见Kafka制作人属性。

记录元数据通道

成功发送结果应发送到的MessageChannel的bean名称；bean必须存在于应用程序上下文中。发送到通道的消息是已发送的消息（转换后，如果有），带有附加标头KafkaHeaders。RECORD_METADATA。标头包含Kafka客户端提供的RecordMetadata对象；它包括在主题中写入记录的分区和偏移量。

结果元数据元数据=SendResultMg。getHeaders（）。get（KafkaHeaders.RECORD\u METADATA，RecordMetadata.class）

失败发送到生产者错误通道（如果配置）；请参阅错误通道。默认值：null

您可以添加一个ServiceActivator来异步使用此通道。