Kafka 开飙了！5分钟，带你体验一把“速度与激情”

约定不等于承诺〃 2023-10-15 01:56 64阅读 0赞

前置条件：你的电脑已经安装 Docker

主要内容：

1.  使用 Docker 安装
2.  使用命令行测试消息队列的功能
3.  zookeeper和kafka可视化管理工具
4.  Java 程序中简单使用Kafka

## 使用 Docker 安装搭建Kafka环境 ##

**单机版**

**下面使用的单机版的Kafka 来作为演示，推荐先搭建单机版的Kafka来学习。**

> 以下使用 Docker 搭建Kafka基本环境来自开源项目：github.com/simplesteph… 。当然，你也可以按照官方提供的来：github.com/wurstmeiste… 。

新建一个名为  
zk-single-kafka-single.yml 的文件，文件内容如下：

运行以下命令即可完成环境搭建(会自动下载并运行一个 zookeeper 和 kafka )

docker-compose -f zk-single-kafka-single.yml up

如果需要停止Kafka相关容器的话，运行以下命令即可：

docker-compose -f zk-single-kafka-single.yml down

## 集群版 ##

> 以下使用 Docker 搭建Kafka基本环境来自开源项目：github.com/simplesteph… 。

新建一个名为  
zk-single-kafka-multiple.yml 的文件，文件内容如下：

version: '2.1'
    
    services:
      zoo1:
        image: zookeeper:3.4.9
        hostname: zoo1
        ports:
          - "2181:2181"
        environment:
            ZOO_MY_ID: 1
            ZOO_PORT: 2181
            ZOO_SERVERS: server.1=zoo1:2888:3888
        volumes:
          - ./zk-single-kafka-multiple/zoo1/data:/data
          - ./zk-single-kafka-multiple/zoo1/datalog:/datalog
    
      kafka1:
        image: confluentinc/cp-kafka:5.4.0
        hostname: kafka1
        ports:
          - "9092:9092"
        environment:
          KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka1:19092,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
          KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
          KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
          KAFKA_BROKER_ID: 1
          KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
        volumes:
          - ./zk-single-kafka-multiple/kafka1/data:/var/lib/kafka/data
        depends_on:
          - zoo1
    
      kafka2:
        image: confluentinc/cp-kafka:5.4.0
        hostname: kafka2
        ports:
          - "9093:9093"
        environment:
          KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka2:19093,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9093
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
          KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
          KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
          KAFKA_BROKER_ID: 2
          KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
        volumes:
          - ./zk-single-kafka-multiple/kafka2/data:/var/lib/kafka/data
        depends_on:
          - zoo1
    
    
      kafka3:
        image: confluentinc/cp-kafka:5.4.0
        hostname: kafka3
        ports:
          - "9094:9094"
        environment:
          KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka3:19094,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9094
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
          KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
          KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
          KAFKA_BROKER_ID: 3
          KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
        volumes:
          - ./zk-single-kafka-multiple/kafka3/data:/var/lib/kafka/data
        depends_on:
          - zoo1

运行以下命令即可完成 1个节点 Zookeeper+3个节点的 Kafka 的环境搭建。

docker-compose -f zk-single-kafka-multiple.yml up

如果需要停止Kafka相关容器的话，运行以下命令即可：

docker-compose -f zk-single-kafka-multiple.yml down

## 使用命令行测试消息的生产和消费 ##

一般情况下我们很少会用到 Kafka 的命令行操作。

**1.进入 Kafka container 内部执行 Kafka 官方自带了一些命令**

docker exec -ti docker_kafka1_1 bash

**2.列出所有 Topic**

root@kafka1:/# kafka-topics --describe --zookeeper zoo1:2181

**3.创建一个 Topic**

root@kafka1:/# kafka-topics --create --topic test --partitions 3 --zookeeper zoo1:2181 --replication-factor 1
    Created topic test.

我们创建了一个名为 test 的 Topic, partition 数为 3, replica 数为 1。

**4.消费者订阅主题**

root@kafka1:/# kafka-console-consumer --bootstrap-server localhost:9092 --topic test
    send hello from console -producer

我们订阅了 名为 test 的 Topic。

**5.生产者向 Topic 发送消息**

root@kafka1:/# kafka-console-producer --broker-list localhost:9092 --topic test
    >send hello from console -producer
    >

我们使用 kafka-console-producer 命令向名为 test 的 Topic 发送了一条消息，消息内容为：“send hello from console -producer”

这个时候，你会发现消费者成功接收到了消息：

root@kafka1:/# kafka-console-consumer --bootstrap-server localhost:9092 --topic test
    send hello from console -producer

## IDEA相关插件推荐 ##

## Zoolytic-Zookeeper tool ##

这是一款 IDEA 提供的 Zookeeper 可视化工具插件，非常好用！ 我们可以通过它：

1.  可视化ZkNodes节点信息
2.  ZkNodes节点管理-添加/删除
3.  编辑zkNodes数据
4.  ......实际使用效果如下：

![ca25394ace5346c6862b610b64769226.png][]![f85cf4d6327c475b9a3812a8d164ddbf_tplv-tt-origin-asy1_5aS05p2hQOS7o-eggeWwj-W9k-Wutg_.image_iz_58558_from_article.pc_detail_x-expires_1693190226_x-signature_L_2FT374MVMQ_2F9FKZo5yaBIm_2F3Q7A_3D][]

使用方法：

1.  打开工具：View->Tool windows->Zoolytic；
2.  点击 “+” 号后在弹出框数据：“127.0.0.1:2181” 连接 zookeeper；
3.  连接之后点击新创建的连接然后点击“+”号旁边的刷新按钮即可！

## Kafkalytic ##

IDEA 提供的 Kafka 可视化管理插件。这个插件为我们提供了下面这写功能：

1.  多个集群支持
2.  主题管理:创建/删除/更改分区
3.  使用正则表达式搜索主题
4.  发布字符串/字节序列化的消息
5.  使用不同的策略消费消息

实际使用效果如下：

![924293d805a04f6fac7225d6860be828.png][]

![17a698e32b9d4da388f1ad1ad56e4cc3_tplv-tt-origin-asy1_5aS05p2hQOS7o-eggeWwj-W9k-Wutg_.image_iz_58558_from_article.pc_detail_x-expires_1693190226_x-signature_N60AdXyE42FHBMdQbTbbopRWi6M_3D][]

使用方法：

1.  打开工具：View->Tool windows->kafkalytic；
2.  点击 “+” 号后在弹出框数据：“127.0.0.1:9092” 连接；

## Java 程序中简单使用Kafka ##

> 代码地址：github.com/Snailclimb/…

**Step 1:新建一个Maven项目**

**Step2: pom.xml 中添加相关依赖**

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

**Step 3：初始化消费者和生产者**

KafkaConstants常量类中定义了Kafka一些常用配置常量。

public class KafkaConstants {
        public static final String BROKER_LIST = "localhost:9092";
        public static final String CLIENT_ID = "client1";
        public static String GROUP_ID_CONFIG="consumerGroup1";
        private KafkaConstants() {
    
        }
    }

ProducerCreator 中有一个 createProducer() 方法方法用于返回一个 KafkaProducer对象

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    /**
     * @author shuang.kou
     */
    public class ProducerCreator {
    
    
        public static Producer<String, String> createProducer() {
            Properties properties = new Properties();
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.BROKER_LIST);
            properties.put(ProducerConfig.CLIENT_ID_CONFIG, KafkaConstants.CLIENT_ID);
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            return new KafkaProducer<>(properties);
        }
    }

ConsumerCreator 中有一个createConsumer() 方法方法用于返回一个 KafkaConsumer 对象

import org.apache.kafka.clients.consumer.Consumer;
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.common.serialization.StringDeserializer;
    
    import java.util.Properties;
    
    public class ConsumerCreator {
    
        public static Consumer<String, String> createConsumer() {
            Properties properties = new Properties();
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.BROKER_LIST);
            properties.put(ConsumerConfig.GROUP_ID_CONFIG, KafkaConstants.GROUP_ID_CONFIG);
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            return new KafkaConsumer<>(properties);
        }
    }

**Step 4：发送和消费消息**

生产者发送消息：

private static final String TOPIC = "test-topic";
    Producer<String, String> producer = ProducerCreator.createProducer();
    ProducerRecord<String, String> record =
     new ProducerRecord<>(TOPIC, "hello, Kafka!");
    try {
     //send message
     RecordMetadata metadata = producer.send(record).get();
     System.out.println("Record sent to partition " + metadata.partition()
                        + " with offset " + metadata.offset());
    } catch (ExecutionException | InterruptedException e) {
     System.out.println("Error in sending record");
     e.printStackTrace();
    }
    producer.close();

消费者消费消息：

Consumer<String, String> consumer = ConsumerCreator.createConsumer();
    // 循环消费消息
    while (true) {
      //subscribe topic and consume message
      consumer.subscribe(Collections.singletonList(TOPIC));
    
      ConsumerRecords<String, String> consumerRecords =
        consumer.poll(Duration.ofMillis(1000));
      for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        System.out.println("Consumer consume message:" + consumerRecord.value());
      }
    }

**Step 5：测试**

运行程序控制台打印出：

Record sent to partition 0 with offset 20
    Consumer consume message:hello, Kafka!

## 开源项目推荐 ##

作者的其他开源项目推荐：

1.  JavaGuide：【Java学习+面试指南】 一份涵盖大部分Java程序员所需要掌握的核心知识。
2.  springboot-guide : 适合新手入门以及有经验的开发人员查阅的 Spring Boot 教程（业余时间维护中，欢迎一起维护）。
3.  programmer-advancement : 我觉得技术人员应该有的一些好习惯！
4.  spring-security-jwt-guide :从零入门 ！Spring Security With JWT（含权限验证）后端部分代码。

[ca25394ace5346c6862b610b64769226.png]: https://img-blog.csdnimg.cn/ca25394ace5346c6862b610b64769226.png
[f85cf4d6327c475b9a3812a8d164ddbf_tplv-tt-origin-asy1_5aS05p2hQOS7o-eggeWwj-W9k-Wutg_.image_iz_58558_from_article.pc_detail_x-expires_1693190226_x-signature_L_2FT374MVMQ_2F9FKZo5yaBIm_2F3Q7A_3D]: https://p3-sign.toutiaoimg.com/pgc-image/f85cf4d6327c475b9a3812a8d164ddbf~tplv-tt-origin-asy1:5aS05p2hQOS7o-eggeWwj-W9k-Wutg==.image?_iz=58558&from=article.pc_detail&x-expires=1693190226&x-signature=L%2FT374MVMQ%2F9FKZo5yaBIm%2F3Q7A%3D
[924293d805a04f6fac7225d6860be828.png]: https://img-blog.csdnimg.cn/924293d805a04f6fac7225d6860be828.png
[17a698e32b9d4da388f1ad1ad56e4cc3_tplv-tt-origin-asy1_5aS05p2hQOS7o-eggeWwj-W9k-Wutg_.image_iz_58558_from_article.pc_detail_x-expires_1693190226_x-signature_N60AdXyE42FHBMdQbTbbopRWi6M_3D]: https://p3-sign.toutiaoimg.com/pgc-image/17a698e32b9d4da388f1ad1ad56e4cc3~tplv-tt-origin-asy1:5aS05p2hQOS7o-eggeWwj-W9k-Wutg==.image?_iz=58558&from=article.pc_detail&x-expires=1693190226&x-signature=N60AdXyE42FHBMdQbTbbopRWi6M%3D