Kafka生产者：如何使用响应式和非阻塞生产者编写onSuccess/onError回调函数

使用响应式和非阻塞生产者编写onSuccess/onError回调函数的解决方法如下所示：

首先，我们需要引入相关的依赖项，包括Kafka客户端和Reactor库：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

<dependency>
    <groupId>io.projectreactor.kafka</groupId>
    <artifactId>reactor-kafka</artifactId>
    <version>1.4.0</version>
</dependency>

接下来，我们可以编写一个生产者类并实现响应式和非阻塞的生产者逻辑。以下是一个示例：

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import reactor.core.publisher.Mono;
import reactor.kafka.sender.KafkaSender;
import reactor.kafka.sender.SenderRecord;
import reactor.kafka.sender.SenderResult;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

public class ReactiveKafkaProducer {

    private KafkaSender<String, String> kafkaSender;

    public ReactiveKafkaProducer() {
        Map<String, Object> producerProps = new HashMap<>();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, "reactive-producer");

        kafkaSender = KafkaSender.create(producerProps);
    }

    public CompletableFuture<RecordMetadata> send(String topic, String key, String value) {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, key, value);
        SenderRecord<String, String, String> senderRecord = new SenderRecord<>(producerRecord, null);

        Mono<SenderResult<String>> senderResultMono = kafkaSender.send(Mono.just(senderRecord));

        CompletableFuture<SenderResult<String>> completableFuture = senderResultMono.toFuture();
        return completableFuture.thenApply(SenderResult::recordMetadata);
    }

    public void close() {
        kafkaSender.close();
    }
}

在上面的代码中，我们创建了一个KafkaSender对象，并设置了相关的Kafka生产者配置。然后，我们定义了一个send方法，该方法接收一个主题、键和值，并返回一个CompletableFuture，该对象在发送完成后会返回RecordMetadata对象。

在send方法中，我们首先创建一个ProducerRecord对象，然后使用它创建一个SenderRecord对象。接下来，我们使用kafkaSender.send方法发送SenderRecord并获取一个Mono<SenderResult>对象。我们将这个Mono对象转换为CompletableFuture，然后使用thenApply方法将SenderResult转换为RecordMetadata。

最后，我们在close方法中关闭kafkaSender对象。

你可以根据自己的需求修改和扩展这个示例。