java 配置kafka

要在Java中使用Kafka，需要进行以下配置步骤：

1. 下载Kafka：从Apache Kafka官网下载最新的Kafka二进制文件。

2. 解压文件：将下载的Kafka压缩文件解压到本地文件系统上。

3. 启动Kafka Server：在终端中进入Kafka目录，使用以下命令启动Kafka Server：

    bin/kafka-server-start.sh config/server.properties

4. 创建Topic：使用以下命令创建一个名为“test”的Topic：

    bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

5. 创建Producer：使用以下Java代码创建一个Producer：

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

    import java.util.Properties;

    public class MyKafkaProducer {
        public static void main(String[] args) {
            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);
            for (int i = 0; i < 10; i++) {
                producer.send(new ProducerRecord<>("test", Integer.toString(i), Integer.toString(i)));
            }

            producer.close();
        }
    }

6. 创建Consumer：使用以下Java代码创建一个Consumer：

    import org.apache.kafka.clients.consumer.*;
    import java.util.*;

    public class MyKafkaConsumer {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("group.id", "test");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Arrays.asList("test"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records)
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }

这些配置步骤将使您能够在Java中使用Kafka。