1、从Kafka的官网下载Kafka安装包：Apache Kafka

我下的是2.8.0

对应pom.xml配置

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

2、下载后解压到C盘

修改路径，在解压后的config目录下（kafka_2.13-2.8.0\config），

①zookeeper.properties中修改zookeeper的路径：dataDir=C:/kafka/kafka_2.13-2.8.0/zookeeper

②server.properties中修改kafka的路径：log.dirs=C:/kafka/kafka_2.13-2.8.0/kafka-logs

3、启动测试：（在指定文件夹路径中写cmd就可以打开了）

①先启动zookeeper：C:/kafka/kafka_2.13-2.8.0 > bin\windows\zookeeper-server-start.bat config\zookeeper.properties

②再启动kafka：C:/kafka/kafka_2.13-2.8.0>bin\windows\kafka-server-start.bat config\server.properties

ps：若显示“此时不应有 \kafka_2.13-2.8.0\bin\windows\../../config/log4j.properties”的报错，说明路径中有空格，把文件放到别的位置就可以了。

4、查看端口9092是否打开

在命令行中输入netstat -an

5、java代码获取kafka信息

public class consumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "IP地址:9092"); //kafka集群broker列表
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "topic"); //指定topic
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            records.forEach(record -> {
                System.out.printf("Offset = %d, Key = %s, Value = %s%n", record.offset(), record.key(), record.value());
            });
        }
    }
}