Kafka生产者onCompletion()与get()方法的区别及Java使用详解

核心区别 #

1. 执行模式 #

• get()：阻塞同步调用，调用后主线程暂停，直到Kafka返回发送结果（成功或失败）才继续执行。
• onCompletion()：异步非阻塞回调，调用send()后主线程立即继续执行，发送结果由Kafka客户端在后台线程触发回调函数处理。

2. 性能与吞吐量 #

• get()：每次发送都会阻塞主线程，无法并行处理消息，在高吞吐场景下会大幅降低生产者性能。
• onCompletion()：主线程无需等待，生产者可继续发送下一条消息，配合默认的batch.size（16384字节）和linger.ms（0），能充分利用批量发送优势，提升吞吐量。

3. 异常处理 #

• get()：会抛出InterruptedException、ExecutionException等检查型异常，必须显式捕获或声明抛出，异常直接暴露在主线程中。
• onCompletion()：异常通过回调参数exception传递，无需在主线程捕获，处理逻辑封装在回调内部，不会打断主线程流程。

4. 使用场景 #

• get()：适合同步发送场景，比如需要立即确认消息发送结果、依赖发送结果执行后续操作的业务（如订单创建后必须确认消息发送成功才返回成功）。
• onCompletion()：适合异步发送场景，比如日志采集、埋点上报等不依赖发送结果的业务，或需要批量高效发送消息的场景。

结合默认属性的Java实现示例 #

默认属性说明：acks=1（broker写入本地日志即确认）、batch.size=16384（批量发送阈值）、max.block.ms=60000（发送阻塞超时时间）、linger.ms=0（消息立即发送，不等待攒批）。

1. 使用get()方法同步获取结果 #

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 KafkaGetSyncExample {
    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");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "order-key", "order-value");
            
            // 调用get()阻塞等待发送结果，默认最多阻塞60秒
            RecordMetadata metadata = producer.send(record).get();
            
            System.out.printf("消息发送成功：主题=%s，分区=%d，偏移量=%d%n",
                    metadata.topic(), metadata.partition(), metadata.offset());
        } catch (Exception e) {
            System.err.println("消息发送失败：" + e.getMessage());
            e.printStackTrace();
        }
    }
}

2. 使用onCompletion()异步处理结果 #

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

public class KafkaCallbackAsyncExample {
    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");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "log-key", "log-value");
            
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        System.out.printf("异步消息发送成功：主题=%s，分区=%d，偏移量=%d%n",
                                metadata.topic(), metadata.partition(), metadata.offset());
                    } else {
                        System.err.println("异步消息发送失败：" + exception.getMessage());
                        exception.printStackTrace();
                    }
                }
            });
            
            System.out.println("主线程继续处理其他业务...");
        }
    }
}

内容的提问来源于stack exchange，提问作者Misubushi