Kafka 入门指南与一键摆设

Kafka 先容

想象一下你正在运营一个大型电商平台，每秒都有成千上万的用户浏览商品、下单、付出，同时后台系统还在记载用户举动、更新库存、处理物流信息。这些海量、连续产生的数据就像奔驰不息的河道，你需要一个强大、可靠且实时的系统来接收、存储并处理这些数据洪流，让差别的业务部门（比如实时保举、库存管理、风险控制、用户举动分析）都能及时获取他们需要的信息。这就是Apache Kafka大展拳脚的舞台。
本质上，Kafka是一个开源的、分布式的流处理平台。它最焦点的能力是作为一个高吞吐量、低延迟、可水平扩展且持久化的发布-订阅消息系统。你可以把它想象成一个超等高效、永不丢失的“消息管道”或“数据中枢”。它的计划哲学就是处理源源不停产生的实时数据流（Data Streams）。
我们看几个详细的例子：

• 构建实时数据管道和流处理： 这是Kafka的看家本领。比如在金融行业，一家证券交易所需要实时处理每秒产生的巨量股票交易订单数据。Kafka作为焦点枢纽，可靠地接收来自各个交易终端发来的订单消息，持久化存储它们。然后，风控系统可以实时订阅这些数据流，毫秒级地检测异常交易模式防止欺诈；交易撮合引擎订阅数据举行实时匹配；同时，另一个消费者大概将这些数据实时推送到数据仓库或Hadoop集群，供分析师举行更深入的历史趋势研究。全部系统都通过Kafka这个统一、可靠的管道获取实时数据，避免了复杂的点对点集成。
  
• 网站运动追踪和日志聚合： 想象一下像滴滴这样的大型应用。每当乘客打开APP、搜刮所在、呼唤车辆、司机接单、车辆移动、行程竣事、付出完成，每一步都会产生一条事故日志。这些日志数目巨大且分散在各个服务器上。Kafka提供了一个中心化的地方，让全部服务器都可以轻松地将这些运动事故（比如“用户A在时间T点击了按钮B”）作为消息发布到特定的主题（Topics）里。下游的各种系统可以按需订阅：实时监控系统订阅这些流来监控APP的实时康健状态和用户举动漏斗；用户画像系统订阅来更新用户的实时偏好和举动轨迹；安全团队订阅来实时检测可疑运动（如异常频繁的登录尝试）。同时，这些日志也可以被消费到像Elasticsearch这样的系统中提供快速的搜刮和可视化，大概到HDFS做长期存储和离线分析。Kafka高效地统一了日志收集的入口。
  
• 物联网数据集成： 在智能工厂或智慧城市项目中，成千上万的传感器（温度、湿度、压力、位置、摄像头图像元数据等）每时每刻都在产生数据。这些传感器装备（或边沿网关）将收罗到的数据发送到Kafka。Kafka依附其高吞吐量和分布式特性，轻松承接这些海量、高速的传感器数据流。实时监控中心订阅这些数据流，可以立即在大屏上展示工厂装备的运行状态或城市的交通流量；猜测性维护系统分析装备传感器数据流，实时判定机器是否大概出现故障；数据湖则消费这些数据，存储起来供后续训练AI模子优化生产流程或城市规划。
  
• 解耦微服务通讯： 在一个由许多小型、独立服务（微服务）构成的当代应用架构中，服务之间需要通讯。Kafka可以作为它们之间的可靠“缓冲带”或“通讯总线”。例如，在一个电商系统中，“订单服务”处理完一个新订单后，它不需要直接调用“库存服务”、“付出服务”和“物流服务”，而是简单地将一条“新订单创建”的消息发布到Kafka的一个主题里。各个相关的服务（库存扣减、付出处理、物流调度）都独立地订阅这个主题。这样，“订单服务”只需要快速把消息发出去就完成任务了，不消等候或关心下游服务何时处理、是否成功（下游服务自己包管消费可靠性），大大进步了系统的整体相应速度、可扩展性和容错性。纵然某个下游服务（如物流服务）暂时宕机，消息也会安全地保存在Kafka中，等它恢复后继续处理。
  
• 事故溯源和变更数据捕捉： 在一些需要精确记载状态变化历史的系统中（如银行焦点系统、审计系统），Kafka可以用来存储全部导致状态变化的事故序列。例如，一个银行账户的每次存款、取款、转账操纵都被记载为一个不可变的事故，发布到Kafka。通过重放这些事故流，可以精确地重建账户在任何历史时刻的状态，提供了强大的审计追踪能力。同时，数据库的变更（CDC - Change Data Capture）也可以通过工具捕捉（如读取数据库的binlog），转换成事故流发布到Kafka，让其他系统能够近乎实时地感知到数据库的变更。
  

总结来说，Kafka的焦点代价在于它能够：

• 可靠地处理海量实时数据流：像一个永不堵塞的高速公路。
  
• 持久化存储数据流：数据可以保留很长时间（几天、几周甚至几个月），允许差别的消费者按自己的节奏重放历史数据。
  
• 连接差别的系统和应用：作为统一、可靠的数据中枢，简化系统架构，实现松耦合。
  
• 支撑实时处理和分析：为需要即时相应的业务场景（监控、风控、保举、告警）提供实时数据源。
  

Kafka角色和流程

Kafka角色

• Producer: Producer即生产者，消息的产生者，是消息的入口。负责发布消息到Kafka broker。
  
• Consumer: 消费者，用于消费消息，即处理消息
  
• Broker：Broker是kafka实例，每个服务器上可以有一个或多个kafka的实例，假设每个broker对应一台服务器，每个kafka集群内的broker都有一个不重复的编号，如: broker-0, broker-1等……
  
• Topic: 消息的主题，可以明白为消息的分类，相当于Redis的Key和ES中的索引，kafka的数据就保存在Topic，在每个broker上都可以创建多个Topic，物理上差别 topic 的消息分开存储在差别的文件夹，逻辑上一个 topic的消息虽然保存于一个或多个broker 上，但用户只需指定消息的Topic即可天生声明数据而不必关心数据存于何处。Topic 在逻辑上对record记载、日志举行分组保存，消费者需要订阅相应的Topic才气满意Topic中的消息。
  
• Consumer group: 每个consumer 属于一个特定的consumer group（可为每个consumer 指定 group name，若不指定 group name 则属于默认的group），同一topic的一条消息只能被同一个consumer group 内的一个consumer 请求，类似于一对一的单播机制，但多个consumer group 可同时请求这一消息，类似于一对多的多播机制
  
• Partition：是物理上的概念，每个topic 分别为一个或多个partition，即一个topic划分为多份创建 topic的可指定 partition 数目。partition的体现情势就是一个一个的文件夹,该文件夹下存储该partition的数据和索引文件，分区的作用还可以实现负载平衡，进步kafka的吞吐量。同一个topic在差别的分区的数据是不重复的,一般Partition数不要超过节点数。留意同一个partition数据是有次序的，但差别partition则是无序的。
  
• Replication: 同样数据的副本，包罗leader和follower的副本数基本于数据安全,发起至少2个,是Kafka的高可靠性的保障。和ES的副本有所差别，Kafka中的副本数包罗主分片数,而ES中的副本数不包罗主分片数
  
• AR：Assigned Replicas，分区中的全部副本的统称，包罗leader和 follower。AR= ISR+ OSR
  
• ISR：In Sync Replicas，全部与leader副本保持同步的副本 follower和leader自己组成的集合，包罗leader和 follower，是AR的子集
  
• OSR：out-of-Sync Replied，全部与leader副本同步不能同步的 follower的集合，是AR的子集
  

Kafka写入消息流程

Kafka 配置文件说明

1. 配置文件说明
3. # 配置文件 ./conf/server.properties 内部说明
5. ******** Server Basics ********
7. # broker 的 id，值为整数，且必须唯一，在一个集群中不能重复
8. broker.id=1
10. ******** Socket Server Settings ********
12. # kafka 监听端口，默认 9092
13. listeners=PLAINTEXT://10.0.0.101:9092
15. # 处理网络请求的线程数量，默认为 3 个
16. num.network.threads=3
18. # 执行磁盘 IO 操作的线程数量，默认为 8 个
19. num.io.threads=8
21. # socket 服务发送数据的缓冲区大小，默认 100kB
22. socket.send.buffer.bytes=102400
24. # socket 服务接受数据的缓冲区大小，默认 100kB
25. socket.receive.buffer.bytes=102400
27. # socket 服务所能接受的一个请求的最大大小，默认为 100M
28. socket.request.max.bytes=104857600
30. ******** Log Basics ********
32. # kafka 存储消息数据的目录
33. log.dirs=../data
35. # 每个 topic 的默认 partition 数量
36. num.partitions=1
38. # 设置默认副本数量为 3。如果 Leader 副本故障，会进行故障自动转移。
39. default.replication.factor=3
41. # 在启动时恢复数据和关闭时刷数据前，每个数据目录使用的线程数量
42. num.recovery.threads.per.data.dir=1
44. ******** Log Flush Policy ********
46. # 消息刷新到磁盘中的消息条数阈值
47. log.flush.interval.messages=10000
49. # 消息刷新到磁盘中的最大时间间隔, 1s
50. log.flush.interval.ms=1000
52. ******************************************************* Log Retention Policy *******************************************************
54. # 日志保留时间，超时会自动删除，默认为 7 天 (168 小时)
55. log.retention.hours=168
57. # 日志保留大小，超出大小会自动删除，默认为 1GB
58. log.retention.bytes=1073741824
60. # 单个日志段文件的大小限制，最大为 1GB，超出后则创建一个新的日志段文件
61. log.segment.bytes=1073741824
63. # 每隔多长时间检测数据是否达到删除条件, 300s (5 分钟)
64. log.retention.check.interval.ms=300000
66. ******** Zookeeper ********
68. # Zookeeper 连接信息，如果是 Zookeeper 集群，则以逗号隔开
69. zookeeper.connect=10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181
71. # 连接 Zookeeper 的超时时间, 6s
72. zookeeper.connection.timeout.ms=6000
74. # 是否允许删除 topic，默认为 false（topic 只会被标记为待删除）。设置为 true 则允许物理删除。
75. delete.topic.enable=true

复制代码

Kafka 集群摆设

versionIPZookeeper3.8.4192.168.80.11-80.33Kafka3.8.0192.168.80.11-80.33 一键安装脚本

每个节点分别执行

1. #!/bin/bash
2. KAFKA_VERSION=3.8.0
3. SCALA_VERSION=2.13
4. KAFKA_URL="https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/${KAFKA_VERSION}/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz"
5. ZK_VERSION=3.8.4
7. ZK_URL="https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-${ZK_VERSION}/apache-zookeeper-${ZK_VERSION}-bin.tar.gz"
9. ZK_INSTALL_DIR=/usr/local/zookeeper
10. KAFKA_INSTALL_DIR=/usr/local/kafka
12. NODE1=192.168.80.11
13. NODE2=192.168.80.12
14. NODE3=192.168.80.13
16. HOST=$(hostname -I | awk '{print $1}')
17. . /etc/os-release
19. print_status() {
20.     local msg=$1
21.     local status=$2
22.     
23.     local GREEN='\033[0;32m'
24.     local RED='\033[0;31m'
25.     local YELLOW='\033[0;33m'
26.     local NC='\033[0m'
27.     
28.     case $status in
29.         success)
30.             color=$GREEN
31.             symbol="[  OK  ]"
32.             ;;
33.         failure)
34.             color=$RED
35.             symbol="[FAILED]"
36.             ;;
37.         warning)
38.             color=$YELLOW
39.             symbol="[WARNING]"
40.             ;;
41.     esac
42.     
43.     printf "%-50s ${color}%s${NC}\n" "$msg" "$symbol"
44. }
46. detect_node_id() {
47.     if [[ $HOST == $NODE1 ]]; then
48.         MYID=1
49.     elif [[ $HOST == $NODE2 ]]; then
50.         MYID=2
51.     elif [[ $HOST == $NODE3 ]]; then
52.         MYID=3
53.     else
54.         while true; do
55.             read -p "输入节点编号 (1-3): " MYID
56.             if [[ $MYID =~ ^[1-3]$ ]]; then
57.                 break
58.             else
59.                 print_status "无效的节点编号" warning
60.             fi
61.         done
62.     fi
63.     print_status "节点ID: $MYID" success
64. }
66. install_java() {
67.     if [[ $ID == 'centos' || $ID == 'rocky' ]]; then
68.         yum -y install java-1.8.0-openjdk-devel
69.     else
70.         apt update
71.         apt install -y openjdk-8-jdk
72.     fi
73.     
74.     if java -version &>/dev/null; then
75.         print_status "Java安装成功" success
76.     else
77.         print_status "Java安装失败" failure
78.         exit 1
79.     fi
80. }
82. setup_zookeeper() {
83.     local src_dir="/usr/local/src"
84.     local pkg_name=$(basename $ZK_URL)
85.     
86.     if [[ ! -f "$src_dir/$pkg_name" ]]; then
87.         wget -P $src_dir $ZK_URL || {
88.             print_status "ZooKeeper下载失败" failure
89.             exit 1
90.         }
91.     fi
92.     
93.     tar xf "$src_dir/$pkg_name" -C /usr/local || {
94.         print_status "解压失败" failure
95.         exit 1
96.     }
97.     
98.     ln -sf /usr/local/apache-zookeeper-*-bin $ZK_INSTALL_DIR
99.     
100.     echo "PATH=$ZK_INSTALL_DIR/bin:\$PATH" > /etc/profile.d/zookeeper.sh
101.     source /etc/profile.d/zookeeper.sh
102.     
103.     mkdir -p $ZK_INSTALL_DIR/data
104.     echo $MYID > $ZK_INSTALL_DIR/data/myid
105.     
106.     cat > $ZK_INSTALL_DIR/conf/zoo.cfg <<EOF
107. tickTime=2000
108. initLimit=10
109. syncLimit=5
110. dataDir=$ZK_INSTALL_DIR/data
111. clientPort=2181
112. maxClientCnxns=128
113. autopurge.snapRetainCount=3
114. autopurge.purgeInterval=24
115. server.1=$NODE1:2888:3888
116. server.2=$NODE2:2888:3888
117. server.3=$NODE3:2888:3888
118. EOF
119.    
120.     cat > /etc/systemd/system/zookeeper.service <<EOF
121. [Unit]
122. Description=ZooKeeper Service
123. After=network.target
125. [Service]
126. Type=forking
127. Environment="JAVA_HOME=$(dirname $(dirname $(readlink -f $(which java))))"
128. ExecStart=$ZK_INSTALL_DIR/bin/zkServer.sh start
129. ExecStop=$ZK_INSTALL_DIR/bin/zkServer.sh stop
130. ExecReload=$ZK_INSTALL_DIR/bin/zkServer.sh restart
131. User=root
132. Group=root
133. Restart=on-failure
134. RestartSec=10
136. [Install]
137. WantedBy=multi-user.target
138. EOF
139.     
140.     systemctl daemon-reload
141.     systemctl enable --now zookeeper.service
142.     
143.     if systemctl is-active --quiet zookeeper.service; then
144.         print_status "ZooKeeper启动成功" success
145.     else
146.         print_status "ZooKeeper启动失败" failure
147.         exit 1
148.     fi
149. }
151. setup_kafka() {
152.     local src_dir="/usr/local/src"
153.     local pkg_name=$(basename $KAFKA_URL)
154.     
155.     if [[ ! -f "$src_dir/$pkg_name" ]]; then
156.         wget -P $src_dir $KAFKA_URL || {
157.             print_status "Kafka下载失败" failure
158.             exit 1
159.         }
160.     fi
161.     
162.     tar xf "$src_dir/$pkg_name" -C /usr/local || {
163.         print_status "解压失败" failure
164.         exit 1
165.     }
166.     
167.     ln -sf /usr/local/kafka_* $KAFKA_INSTALL_DIR
168.     
169.     echo "PATH=$KAFKA_INSTALL_DIR/bin:\$PATH" > /etc/profile.d/kafka.sh
170.     source /etc/profile.d/kafka.sh
171.     
172.     mkdir -p $KAFKA_INSTALL_DIR/data
173.     
174.     cat > $KAFKA_INSTALL_DIR/config/server.properties <<EOF
175. broker.id=$MYID
176. listeners=PLAINTEXT://$HOST:9092
177. log.dirs=$KAFKA_INSTALL_DIR/data
178. num.partitions=1
179. log.retention.hours=168
180. zookeeper.connect=$NODE1:2181,$NODE2:2181,$NODE3:2181
181. zookeeper.connection.timeout.ms=6000
182. EOF
183.     
184.     cat > /etc/systemd/system/kafka.service <<EOF
185. [Unit]
186. Description=Apache Kafka
187. After=network.target zookeeper.service
189. [Service]
190. Type=simple
191. Environment="JAVA_HOME=$(dirname $(dirname $(readlink -f $(which java))))"
192. ExecStart=$KAFKA_INSTALL_DIR/bin/kafka-server-start.sh $KAFKA_INSTALL_DIR/config/server.properties
193. ExecStop=/bin/kill -TERM \$MAINPID
194. Restart=on-failure
195. RestartSec=20
197. [Install]
198. WantedBy=multi-user.target
199. EOF
200.     
201.     systemctl daemon-reload
202.     systemctl enable --now kafka.service
203.     
204.     if systemctl is-active --quiet kafka.service; then
205.         print_status "Kafka启动成功" success
206.     else
207.         print_status "Kafka启动失败" failure
208.         exit 1
209.     fi
210. }
212. detect_node_id
213. install_java
214. setup_zookeeper
215. setup_kafka
216. print_status "安装完成" success

复制代码

检察Kafka注册环境

创建Topic

1. /usr/local/kafka/bin/kafka-topics.sh --create --topic DBA --bootstrap-server 192.168.80.11:9092 --partitions 3 --replication-factor 2

复制代码

获取全部 Topic

1. /usr/local/kafka/bin/kafka-topics.sh --list --bootstrap-server 192.168.80.11:9092

复制代码

获取Topic副详情

1. /usr/local/kafka/bin/kafka-topics.sh --describe --bootstrap-server 192.168.80.11:9092

复制代码

Kafka 生产者产出数据

1. /usr/local/zookeeper/bin# /usr/local/kafka/bin/kafka-console-producer.sh --broker-list 192.168.80.11:9092 --topic DBA

复制代码

Kafka 消费者消费数据

1. /usr/local/kafka/bin/kafka-console-consumer.sh --topic DBA --bootstrap-server 192.168.80.22:9092 --from-beginning

复制代码

Kafka在Zookeeper中存储布局