kafka 批量写入

对于 Kafka 的批量写入，可以使用 Kafka 生产者 API 提供的批量发送功能来实现。以下是一个示例代码，演示如何使用 Kafka 生产者批量写入消息：

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class KafkaBatchProducerExample {

    public static void main(String[] args) {

        // Kafka 服务器地址
        String bootstrapServers = "localhost:9092";

        // 配置生产者属性
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 创建 Kafka 生产者
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        // 创建消息列表
        List<ProducerRecord<String, String>> records = new ArrayList<>();

        // 添加消息到列表
        records.add(new ProducerRecord<>("topic1", "message1"));
        records.add(new ProducerRecord<>("topic1", "message2"));
        records.add(new ProducerRecord<>("topic1", "message3"));

        // 批量发送消息
        producer.send(records, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception == null) {
                    System.out.println("消息发送成功：" + metadata.topic() + "-" + metadata.partition() + "-" + metadata.offset());
                } else {
                    System.out.println("消息发送失败：" + exception.getMessage());
                }
            }
        });

        // 关闭 Kafka 生产者
        producer.close();
    }
}

在这个示例中，我们首先创建了一个 Kafka 生产者，并设置了服务器地址和序列化器。然后，我们创建了一个消息列表，并将要发送的消息添加到列表中。最后，通过调用 `producer.send()` 方法来批量发送消息。在发送完成后，可以通过回调函数来处理发送结果。

需要注意的是，在实际使用中，你需要根据你的 Kafka 集群和消息的数据类型进行相应的配置和处理。这只是一个简单的示例，供参考。