为啥搭建Kafka需要zookeeper

Posted 2023-05-04

参考技术A spring boot配置开发模式为什么启动会有错误日志采集。线上数据一般主要是落地文件或者通过socket传输给另外一个系统。这种情况下，你很难推动线上应用或服务去修改接口，直接向kafka里写数据。这时候你可能就需要flume这样的系统帮你去做传输。对于数量级别，做过单机upd的flume source的配置，100+M/s数据量，10w qps flume就开始大量丢包。因此我们在搭建系统时，抛弃了flume，自己研发了一套传输系统。但flume设计的source-channel-sink模式还是比较好的，我们在开发系统时无耻的也抄袭了这种方式。

Kafka单机搭建

1、zookeeper搭建

　　Kafka集群依赖zookeeper，需要提前搭建好zookeeper

　　单机模式（7步）(集群模式进阶请移步：http://blog.51cto.com/nileader/795230)

 Step1：

cd /usr/local/software 
wget http://download.oracle.com/otn-pub/java/jdk/8u161-b12/2f38c3b165be4555a1fa6e98c45e0808/jdk-8u161-linux-x64.tar.gz

配置JAVA环境。检验方法：执行java –version和javac –version命令。

Step2：下载并解压zookeeper。

链接：http://mirror.bjtu.edu.cn/apache/zookeeper/zookeeper-3.4.3/，（更多版本：http://dwz.cn/37HGI

 

）最终生成目录类似结构：/home/admin/taokeeper/zookeeper-3.4.3/bin

Step3：重命名 zoo_sample.cfg文件

 mv /home/admin/taokeeper/zookeeper-3.4.3/conf/zoo_sample.cfg  zoo.cfg

 Step4：vi zoo.cfg，修改

dataDir=/home/admin/taokeeper/zookeeper-3.4.3/data

Step5：创建数据目录：mkdir /home/admin/taokeeper/zookeeper-3.4.3/data

1. mkdir  /home/admin/taokeeper/zookeeper-3.4.3/data 

Step6：启动zookeeper：执行

/home/admin/taokeeper/zookeeper-3.4.3/bin/zkServer.sh start

Step7：检测是否成功启动：执行

/home/admin/taokeeper/zookeeper-3.4.3/bin/zkCli.sh 或 echo stat|nc localhost 2181

 

 

2、下载Kafka

    下载地址：http://kafka.apache.org/downloads.html

# mkdir -p /usr/local/software
# cd /usr/local/software
# wget http://mirror.bit.edu.cn/apache/kafka/1.0.0/kafka-1.0.0-src.tgz
# wget http://mirror.bit.edu.cn/apache/kafka/1.0.0/kafka_2.12-1.0.0.tgz

   kafka_2.12-1.0.0.tgz    //其中2.12-1.0.0为Scala的版本，kafka-1.0.0-src.tgz为kafka版本

 

3、解压

# tar zxf kafka_2.12-1.0.0.tgz -C /usr/local/
# cd /usr/local/
# mv kafka_2.12-1.0.0/ kafka/

4、配置

# vi /usr/local/kafka/config/server.properties

# broker的ID，集群中每个broker ID不可相同
broker.id=0
# 监听器，端口号和port一致即可
listeners=PLAINTEXT://:9092
# Broker的监听端口
port=9092
# 必须填写当前服务器IP地址
host.name=10.10.6.225
# 必须填写当前服务器IP地址
advertised.host.name=10.10.6.225
# 暂未配置集群
#zookeeper.connect=zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181

 

5、配置Kafka的环境变量

# vi /etc/profile
　　export KAFKA_HOME=/usr/local/kafka
　　export PATH=$PATH:$KAFKA_HOME/bin
# source /etc/profile

6、启动Kafka

# kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

　　官方推荐启动方式：

# /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &

但这种方式退出shell后会自动断开

7、验证

# jps
    2256 Jps
    2237 Kafka

看到Kafka的进程，说明Kafka已经启动

8、创建topic

    创建名为test，partitions为3，replication为3的topic

# kafka-topics.sh --create --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --partitions 3 --replication-factor 3 --topic test

    查看topic状态

# kafka-topics.sh --describe --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --topic test
　　Topic:test PartitionCount:3 ReplicationFactor:3 Configs:
　　Topic: test Partition: 0 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2
　　Topic: test Partition: 1 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
　　Topic: test Partition: 2 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1

 　　状态说明：test有三个分区分别为1、2、3，分区0的leader是3（broker.id），分区0有三个副本，并且状态都为lsr（ln-sync，表示可以参加选举成为leader）。

    删除topic

    在config/server.properties中加入delete.topic.enable=true并重启服务，在执行如下命令

# kafka-topics.sh --delete --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --topic test

9、测试使用Kafka

    发送消息

# kafka-console-producer.sh --broker-list zookeeper-01:9092,zookeeper-02:9092,zookeeper-03:9092 --topic test
输入以下信息：
　　This is a message
　　This is another message

    接收消息

# kafka-console-consumer.sh --zookeeper zookeeper-01:2181,zookeeper-02:2181,zookeeper-03:2181 --topic test --from-beginning    //--from-beginning 表示从开始第一个消息开始接收

    若看到上输入的信息说明已经搭建成功。