使用Kafka Producer API以实现高吞吐和低延迟。可以使用以下两种方法：

方法一：使用异步发送来提高吞吐量

在异步模式下，生产者将立即返回，而不等待服务器的确认。这允许生产者快速发送下一个消息，从而提高吞吐量。以下是使用异步发送的样例代码。

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

public class AsyncProducerExample {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer producer = new KafkaProducer<>(props);

        for (int i = 0; i < 1000; i++) {
            ProducerRecord record = new ProducerRecord<>("topic1", "key" + i, "value" + i);
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        exception.printStackTrace();
                    } else {
                        System.out.printf("sent record(key=%s value=%s) " +
                                "meta(partition=%d, offset=%d) %n",
                                record.key(), record.value(), metadata.partition(),
                                metadata.offset());
                    }
                }
            });
        }

        producer.close();
    }
}

方法二：调整生产者的批量设置来减少延迟

在这种模式下，你的生产者将要等待一段时间来