windows系统kafka小白入门篇——下载安装，环境配置，入门代码书写

目录

1. kafka 下载

2. 修改配置文件

2.1 文件夹内容

2.2 创建一个 data 空文件夹

2.3 修改 zookeeper.properties 配置文件

2.4 修改 server.properties 配置文件

2.5 创建 &#d" windows脚本文件

2.6 创建 &#d" windows脚本文件

3. 启动 kafka

4. 创建主题+生产者消费者演示

4.1 创建 topic 主题

4.2 命令行创建生产者

4.3 命令行创建消费者

4.4 生产者发送消息供消费者消费

5. demo 书写

5.1 创建项目

5.2 引入依赖

 5.3 生产者测试类代码书写

5.4 消费者测试类代码书写

---

1. kafka 下载

本片是小白入门篇，所以我们不以Linux操作系统为例，选择大多数小白都用的windows。

ksfka下载链接如下，点击链接进入官网即可下载

温馨提示：JDK版本至少需要1.8，高版本也可兼容；

Apache Kafkakafka.apache/downloads本篇中以kafka_2.1.3-3.6.1版本为例，直接点击对应的版本下载即可，tgz包就类似于我们常见的zip，下载完成之后解压即可。

下载完毕，我们就可以解压得到 kafka 了

 解压之后就可以得到 kafka 文件了

2. 修改配置文件

2.1 文件夹内容

打开文件夹后可以发现内部含有bin文件夹，config配置夹，libs依赖夹等，和JDK，maven 问价夹的格式如出一辙；

2.2 创建一个 data 空文件夹

后续需要用来存放日志文件，只要创建完成就可以了，kafka启动后会自动生成日志文件；

2.3 修改 zookeeper.properties 配置文件

我们点击进入config文件夹，找到 zookeeper.properties 配置文件，双击进行修改，

然后，我们找到 dataDir ，将它的值修改为我们刚才创建的 data 文件的路径，还要注意一点，在后面还要多加一个 "/zk"，因为一会还要配置 server.properties ，所以要用将她们两个区分开，

2.4 修改 server.properties 配置文件

和刚才一样，我们双击修改 "server.properties" 配置文件

我们修改 log.dirs 的值为刚才创建的 data 文件夹的路径，在路径末尾再添加上 "/kafka" ，用来和刚才的zk做区分，kafka 文件夹用来存放kafka的日志文件，zk 文件夹用来存放zoopeeper的日志文件；

2.5 创建 &#d" windows脚本文件

  以记事本的方式打开，然后加入下面这句话，

这句话的含义就是启动 Zookeeper ，并且启动文件为 "zookeeper.properties" ；

call bin/windows/zookeeper-server-start.bat config/zookeeper.properties

2.6 创建 &#d" windows脚本文件

仍然以记事本的方式打开，然后加入下面这句话，

这句话的含义就是启动 kafka ，并且启动文件为 "server.properties" ；

call bin/windows/kafka-server-start.bat config/server.properties

此时，我们的 kafka 文件夹中就多了我们刚刚创建的 data 文件夹&#d 脚本文件&#d 脚本文件；

3. 启动 kafka

经过第二部的配置，现在一切都已经准备就绪，我们只需要双击 zk.cmd 和 d 脚本文件启动kafka；

这里需要注意一点，必须先启动双击 zk.cmd 启动 zookeeper，

再双击 d 启动 kafka，关闭的时候，需要先关闭 kafka，再关闭 zookeeper ；

4. 创建主题+生产者消费者演示

4.1 创建 topic 主题

我们来到 bin 文件夹下的 windows 文件夹，打开 cmd 命令窗口，运行下方命令

# --bootstrap-server localhost:9092  配置服务器连接，此处为本机，9092为kafka默认端口号
# --topic test  创建topic主题，主题名称为 test
# --create      创建topic主题命令
kafka-topics.bat --bootstrap-server localhost:9092 --topic test --create     

4.2 命令行创建生产者

仍然是在 windows 文件夹下新建一个命令窗口，刚才我们已经创建出了名为 "topic" 的主题，现在运行如下命令启动脚本文件创建生产者连接上我们的 topic 主题

# 运行 kafka-console-producer.bat 脚本创建生产者连接本机9092端口名为 test 的主题
kafka-console-producer.bat --bootstrap-server localhost:9092 --topic test

 运行如下所示，会出现一个小箭头，就说明我们链接主题成功，我们生产者发布的主题都会发送到 topic 主题中供消费者去消费使用；

4.3 命令行创建消费者

# 运行 kafka-console-consumer.bat 脚本创建消费者连接本机9092端口名为 test 的主题
windows>kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test

运行成功不会有任何显示， 

4.4 生产者发送消息供消费者消费

如下图所示，我在生产者命令窗口输入 "hello kafka"，点击回车，我们就可以在消费者中命令窗口中看到发送过来的 "hello kafka" 消息

5. demo 书写

5.1 创建项目

5.2 引入依赖

在 maven 项目的 l 文件中加入下方依赖，

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

 5.3 生产者测试类代码书写

我们随便创建一个类即可，名字随意取，代码逻辑备有注释；

public class KafkaProducerTest {
    public static void main(String[] args) {
        // TODO 创建配置对象
        // 创建生产者对象又分为两步
            // 1. 创建配置对象集合
        Map<String, Object> configMap = new HashMap<>();
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
            // 2. 配置数据 Key-Value 的序列化方式
        configMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, Name());
        configMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,Name());

        // TODO 创建生产者对象
        // 将配置对象集合作为参数传入
        // 返回值就是生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(configMap);

        // TODO 创建数据
        // 自己定义一个数据，传入三个参数，第一个参数为主题，第二个参数为数据的key,第三个参数为数据的value
        ProducerRecord<String,String> record = new ProducerRecord<>("test","first","hello kafka");

        // TODO 发送数据
        kafkaProducer.send(record);

        // TODO 关闭生产者对象
        kafkaProducer.close();
    }
}

5.4 消费者测试类代码书写

public class KafkaConsumerTest {
    public static void main(String[] args) {
        // TODO 创建消费者配置对象
        // 创建消费者配置对象集合
        HashMap<String, Object> consumerConfig = new HashMap<>();
        consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Name());
        consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,Name());
        // TODO 创建消费者对象
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(consumerConfig);

        // TODO 订阅消费主体，主题名称为 test
        consumer.subscribe(Collections.singletonList("test"));

        // TODO 从主题中获取数据消费
        final ConsumerRecords<String, String> datas = consumer.poll(100);

        for (ConsumerRecord<String,String> data : datas){
            System.out.println(data);
        }

        // TODO 关闭消费者对象
        consumer.close();
    }
}