Kafka集群搭建

以你之姓@ 2022-05-09 05:20 291阅读 0赞

一，环境准备

\* Zookeeper单点/集群服务（演示单点使用单点， 演示集群使用集群）

\* kafka安装包

-- kafka\_2.11-1.1.0.tgz

\* 自己玩， 基于虚拟机搭建

二，Kafka单点搭建

1，安装Zookeeper安装包，可以参考之前Zookeeper集群搭建，不做赘述

2，上传解压kafka安装包到指定目录

tar -zxvf kafka_2.11-1.1.0.tgz -C /usr/develop/

3，修改./conf/server.properties配置文件，修改zookeeper链接地址为真实地址

zookeeper.connect=192.168.91.131:2181

4，启动zookeeper服务，启动前切记关闭防火墙或者开放服务端口（zk：2181；kafka：9092）

// 关闭防火墙
    service iptables stop
    // 启动zookeeper服务
    sh ./bin/zkServer.sh start  
    // 查看zookeeper服务状态
    sh ./bin/zkServer.sh status

5，kafka关联server.properties配置信息启动，&表示后台启动

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

![70][]

6，查看kafka启动状态，存在进程，说明启动成功

ps -ef | grep kafka

![70 1][]

7，添加topic，下列命令表示在zookeeper服务上添加名称为“standloneTest”的topic

\* --partitions 2 ：表示该话题存在两个分区，不指定默认使用（num.partitions配置信息），分区数用于限制对Consumer并行数的限制，Consumer并行数不能大于partition分区数，若超出则分配完分区后，其他Consumer轮空

\* --replication-factor 1 ： 表示每个分区存在一个副本，不指定默认使用（default.replication.factor），具体含义下一篇分析

sh kafka-topics.sh --create --zookeeper 192.168.91.131:2181 --replication-factor 1 --partitions 2 --topic standloneTest

8，kafka日志文件查看节点状态

\* 日志文件路径，在server.properties中配置，下列演示为默认路径演示

log.dirs=/tmp/kafka-logs

\* 查看日志文件

![70 2][]

\* 从上图可以看出，话题standloneTest创建成功，且成功构造出两个分区-0和分区-1

9，zookeeper节点查看topic信息

![70 3][]

\* 从zookeeper节点可以看出， 跟节点下创建了需要支撑kafka的一系列节点，

\* /brokers节点下存在当前kafka节点的topic信息集合，

\* 进入/brokers/topics节点，可以看到该节点上存储上的所有topic，进入创建的topic ： standardTest

\* 进入/brokers/tipics/standardTest节点，可以看到当前topic主题的分区信息，该topic分为两个分区，则节点下有0和1两个分区，其中展示的0和1每个分区，表示该分区在该broker节点下的副本，用于集群中数据同步

\* 获取分区状态信息，如下，具体分区详情信息在集群搭建中再做详解

![70 4][]

10，kafka常用命令

//kafka后台启动命令
    sh ./bin/kafka-server-start.sh -daemon ../config/server.properties
    // kafka停止命令，
    sh ./bin/kafka-server-stop.sh
    // 创建topic命令
    sh ./bin/kafka-topics.sh --create --zookeeper 192.168.91.131:2181 --replication-factor 1 --partitions 2 --topic standloneTest
    // 查看topic列表
    sh ./bin/kafka-topics.sh --list --zookeeper 192.168.91.131:2181
    // 修改topic 
    sh ./bin/kafka-topics.sh --alert --zookeeper 192.168.91.131:2181 --partitions 1 --topic myTest
    // 查看topic信息
    sh ./bin/kafka-topics.sh --describe --zookeeper 192.168.91.131:2181 --topic myTest
    // 删除topic
    sh ./bin/kafka-topics.sh --delete --zookeeper 192.168.91.131:2181 --topic myTest
    // 控制台生产者
    sh ./bin/kafka-console-producer.sh --broker-list 192.168.91.131:9092 --topic myTest
    // 控制台消费者
    sh ./bin/kafka-console-consumer.sh --zookeeper 192.168.91.131:2181 --from-beginning --topic myTest

三，Kafka集群搭建

1，环境准备

\* zookeeper集群环境

\* 三台虚拟机作为服务器搭载kafka

2，修改./conf/server.properties配置文件

\* 修改zookeeper链接

zookeeper.connect=192.168.91.128:2181,192.168.91.129:2181,192.168.91.130:2181

\* 修改broker.id（默认值为0），三台broker.id不一致，

broker.id=1
    broker.id=2
    broker.id=3

\* 修改listeners，修改为当前IP

listeners=PLAINTEXT://192.168.91.128:9092

3，逐节点启动zookeeper集群和kafka集群，链接zookeeper查询节点信息，三个broker在注册中心全部注册成功

![70 5][]

4，基于集群模式创建topic，创建三个分区，每个分区上创建三个副本。保证每个分区的Leader副本可以分布在三台机器上，并且在另外两台机器上分别创建一个Follower副本

sh kafka-topics.sh --create --zookeeper 192.168.91.129:2181,192.168.91.129:2181,192.168.91.130:2181 --replication-factor 3 --partitions 3 --topic clusterTest

5，在日志文件查看分区副本信息，可以看到该服务器下有三个分区的信息，其中一个分区的副本信息为三个副本信息的Leader副本，其余两个分区的副本信息为该分片的Follower副本；副本存在目的主要用于实现数据同步，且副本详情信息存储在zookeeper节点

![70 6][]

6，在zookeeper客户端查看副本详情

\* 查看副本信息详情命令

get /brokers/topics/clusterTest/partitions/0/state

![70 7][]

\* 节点信息分析

-- leader ： 标识当前分区的Leader副本所在的broker.id

-- leader\_epoch ： 同zookeeper集群leader选举的epoch概念，标识当前Leader的版本号

-- isr ： 当前集群下的分片副本信息集合，随着集群信息变化，该isr值动态变更

-- Leader副本所在broker宕机后，会触发Leader重新选举，leader\_epoch加1

四，Java连接Kafka集群进行消息处理

1，服务端发送消息

public class KafkaProducerDemo extends Thread {
    
        private final KafkaProducer<Integer, String> producer;
    
        private final String topic;
    
        public KafkaProducerDemo(String topic) {
            Properties properties = new Properties();
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                    "192.168.91.129:9092,192.168.91.129:9092,192.168.91.130:9092");
            properties.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaProducerDemo");
            properties.put(ProducerConfig.ACKS_CONFIG, "-1");
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.IntegerSerializer");
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringSerializer");
            producer = new KafkaProducer<Integer, String>(properties);
            this.topic = topic;
        }
    
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                String message = "producer message : " + i;
                System.out.println(message);
                producer.send(new ProducerRecord<Integer, String>(topic, message));
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) {
            KafkaProducerDemo producerDemo = new KafkaProducerDemo("clusterTest");
            producerDemo.start();
        }
    
    }

![70 8][]

2，客户端消费消息

public class KafkaConsumerDemo extends Thread {
    
        private final KafkaConsumer<String, Integer> kafkaConsumer;
    
        public KafkaConsumerDemo(String topic) {
            Properties properties = new Properties();
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                    "192.168.91.129:9092,192.168.91.129:9092,192.168.91.130:9092");
            properties.put(ConsumerConfig.GROUP_ID_CONFIG, "KafkaConsumerDemo");
            properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
            properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.IntegerDeserializer");
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringDeserializer");
            // 从第一个顺序读取
            properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            kafkaConsumer = new KafkaConsumer<String, Integer>(properties);
            kafkaConsumer.subscribe(Collections.singleton(topic));
        }
    
        @Override
        public void run() {
            while (true) {
                ConsumerRecords<String, Integer> consumerRecords = kafkaConsumer.poll(1000);
                for (ConsumerRecord record : consumerRecords) {
                    System.out.println("consumer receive : " + record.value());
                }
            }
        }
    
        public static void main(String[] args) {
            KafkaConsumerDemo consumerDemo = new KafkaConsumerDemo("clusterTest");
            consumerDemo.start();
        }
    }

![70 9][]

[70]: /images/20220509/52e291d50194465fa3c164bbc3405ca1.png
[70 1]: /images/20220509/c2539f667e0247e9a4b8edb3b3507528.png
[70 2]: /images/20220509/b431b5e81b484300ac78b0dc962e9a48.png
[70 3]: /images/20220509/067f3e41737e477e86aaf57a519e9ceb.png
[70 4]: /images/20220509/31de12d041b84fe4baf91daee55628b0.png
[70 5]: /images/20220509/8cc4829dcd6e48f3a4ed999e0e343483.png
[70 6]: /images/20220509/3dec0fe7ddd04c83864515e94a145762.png
[70 7]: /images/20220509/05f339234e8845fc9d8c0065bc3a070f.png
[70 8]: /images/20220509/e9a3fc2d1e904a0d820e37ca37723497.png
[70 9]: /images/20220509/33b0cd46af96430bbe49f66019a531e2.png