kafka同步发送一条一条

Kafka是一个高性能的消息队列系统，它的设计目标是实现高吞吐量、低延迟的消息传递，并且保证数据的可靠性。对于Kafka的消息发送，通常有两种方式：同步发送和异步发送。本文将介绍如何使用Kafka同步发送一条一条消息。

一、什么是同步发送？ Kafka提供了两种消息发送方式：同步发送和异步发送。同步发送就是发送一条消息后等待Kafka响应，直到Kafka确认已经接收到了该消息，才会继续发送下一条消息。同步发送的优点是确保消息的可靠性，缺点是发送效率较低，尤其是在消息量较大时。

二、同步发送的实现过程 对于同步发送消息，可以使用Kafka提供的生产者API中的send()方法。该方法会返回一个Future对象，通过判断该对象的返回结果来确定消息发送是否成功。

1.创建一个Kafka生产者对象，设置必要的配置信息，如生产者ID、Kafka集群地址等。

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

Producer<String, String> producer = new KafkaProducer<>(props);

2.使用send()方法发送消息，当Future对象的get()方法返回成功时，表示消息发送成功。

String topic = "test-topic";
String key = "key1";
String value = "value1";
ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
Future<RecordMetadata> future = producer.send(record);
try {
    RecordMetadata metadata = future.get();
    System.out.printf("Message send to partition %d, offset %d%n", metadata.partition(), metadata.offset());
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

3.关闭Kafka生产者对象

producer.close();

三、同步发送的代码示例 下面是一个完整的同步发送消息的示例代码：

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent