4. Kafka 生产者同步与异步发送模式对比

同步与异步发送的核心差异

Kafka 生产者的 send() 方法返回一个 Future<RecordMetadata>。同步发送的本质是立即调用 future.get() 阻塞等待 Broker 确认；异步发送则注册 Callback，由 Sender 线程在响应返回后回调处理。两者共用同一套缓冲与批处理机制，差异集中在调用线程是否阻塞以及异常暴露的位置。

同步发送实现与验证

同步发送适合需要严格确认每条消息写入结果的场景，例如金融交易流水、配置变更通知。以下代码向 order-events 主题发送 100 条消息，每条都等待 Broker 确认。

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class SyncProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all");
        props.put("retries", 3);
        props.put("linger.ms", "0");

        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            long start = System.currentTimeMillis();
            for (int i = 0; i < 100; i++) {
                ProducerRecord<String, String> record =
                    new ProducerRecord<>("order-events", "key-" + i, "order-" + i);
                try {
                    RecordMetadata meta = producer.send(record).get();
                    System.out.printf("offset=%d partition=%d%n",
                        meta.offset(), meta.partition());
                } catch (Exception e) {
                    System.err.println("发送失败: " + e.getMessage());
                }
            }
            long elapsed = System.currentTimeMillis() - start;
            System.out.printf("同步发送 100 条耗时: %d ms, 吞吐: %.1f msg/s%n",
                elapsed, 100000.0 / elapsed);
        }
    }
}

异步发送实现与验证

异步发送适合高吞吐、可容忍少量重试或丢失的场景，例如日志采集、埋点上报。以下代码发送同样 100 条消息，但通过 Callback 处理结果。

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;

public class AsyncProducer {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "1");
        props.put("retries", 3);
        props.put("linger.ms", "20");
        props.put("batch.size", "16384");

        AtomicInteger success = new AtomicInteger();
        AtomicInteger failure = new AtomicInteger();

        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            long start = System.currentTimeMillis();
            for (int i = 0; i < 100; i++) {
                ProducerRecord<String, String> record =
                    new ProducerRecord<>("order-events", "key-" + i, "order-" + i);
                producer.send(record, (metadata, exception) -> {
                    if (exception == null) {
                        success.incrementAndGet();
                    } else {
                        failure.incrementAndGet();
                        System.err.println("异步发送失败: " + exception.getMessage());
                    }
                });
            }
            producer.flush();
            long elapsed = System.currentTimeMillis() - start;
            System.out.printf("异步发送 100 条耗时: %d ms, 成功: %d, 失败: %d%n",
                elapsed, success.get(), failure.get());
        }
    }
}

对比实验与选型依据

在同一台机器、同一主题（3 分区、acks=all）上分别运行上述两个程序，记录耗时。以下为教学示例的预期量级（实际值受硬件和网络影响）：