java 将数据写入kafka 详细代码

以下是使用Java将数据写入Kafka的详细代码：

1. 引入Kafka相关依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.7.0</version>
</dependency>

2. 创建Kafka生产者配置

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092"); // Kafka集群地址
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"); // key序列化类
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // value序列化类

3. 创建Kafka生产者实例

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

4. 构造消息对象

String topic = "test"; // 主题名称
String key = "key"; // 消息键
String value = "Hello, Kafka!"; // 消息值
ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

5. 发送消息

producer.send(record);

6. 关闭Kafka生产者

producer.close();

完整代码如下：

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class KafkaProducerDemo {

    public static void main(String[] args) {
        // Kafka生产者配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // Kafka集群地址
        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"); // key序列化类
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // value序列化类

        // 创建Kafka生产者实例
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 构造消息对象
        String topic = "test"; // 主题名称
        String key = "key"; // 消息键
        String value = "Hello, Kafka!"; // 消息值
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        // 发送消息
        producer.send(record);

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

注意：在实际应用中，我们可以将Kafka生产者封装成一个工具类，提供发送消息的方法，方便其他地方调用。