Kafka 基础

2023-03-08

MiddleWare › Kafka

Kafka

一、概述

1. Kafka 定义

Kafka 是一个分布式的基于发布/订阅模式的消息队列（Message Queue），主要应用于大数据实时处理领域
Kafka 是一个开源的分布式事件流平台（Event Streaming Platform），被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用

使用案例：

网站在前端埋点记录用户操作行为，发送到日志服务器

可通过 Flume 时刻监控日志变化，发送到 Hadoop 等大数据平台进行数据分析

Hadoop 正常读写速度受硬盘影响，约 100M/s。平常 Flume 采集速度小于 100M/s，但在特定场景，其速度会大于 200M/s。为避免造成阻塞，在中间引入 Kafka 集群作为缓冲

2. 消息队列

目前企业中比较常见的消息队列产品主要有 Kafka、ActiveMQ、RabbitMQ、RocketMQ 等
在大数据场景主要采用 Kafka 作为消息队列。在 JavaEE 开发中主要采用 ActiveMQ、RabbitMQ、RocketMQ 等

(1) 传统消息队列的应用场景

缓存/消峰：控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况（如双十一秒杀）
解耦：允许独立的扩展或修改队列两端的处理过程，只要确保它们遵守同样的接口约束（如厂家、超市和顾客）
异步通信：允许用户把一个消息放入队列，但并不立即处理它，而是在需要的时候再去处理（如登录后发送短信通知）

(2) 消息队列的两种模式

点对点模式：消费者主动拉取数据，消息收到后清除消息（一对一）
发布/订阅模式：可以有多个 topic 主题，消费者消费数据之后，不删除数据，且每个消费者相互独立，都可以消费到数据
- 消息的发布者不会将消息直接发送给特定的订阅者，而是将发布的消息分为不同的类别，订阅者只接收感兴趣的消息

mq-mode

3. Kafka 基础架构

kafka-overview

Producer：消息生产者，向 Kafka Broker 发消息的客户端
Consumer：消息消费者，向 Kafka Broker 取消息的客户端
Consumer Group（CG）：消费者组，由多个 consumer 组成
Broker：一台 Kafka 服务器就是一个 Broker
Topic：可以理解为一个队列
Partition：为了实现扩展性，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列，可以分布在不同的 Broker 上
Replica：副本。一个 topic 的每个 partition 都有若干个副本，一个 Leader 和若干个 Follower
Leader：主副本，生产者、消费者只针对 Leader 副本操作
Follower：从副本，实时从 Leader 同步数据，当 Leader 发生故障时，某个 Follower 会成为新的 Leader

二、集群部署

1. 集群规划

hadoop100	hadoop101	hadoop102
ZK、Kafka	ZK、Kafka	ZK、Kafka

2. ZooKeeper 集群部署

下载地址：Apache ZooKeeper

## 准备资源
$ cd /opt/src
$ tar -zxvf apache-zookeeper-3.7.0-bin.tar.gz -C /opt/
$ cd /opt
$ mv apache-zookeeper-3.7.0-bin/ zookeeper
$ cd zookeeper

## 配置ZooKeeper
$ mv conf/zoo_sample.cfg conf/zoo.cfg
$ vim conf/zoo.cfg
dataDir=/opt/zookeeper/data
server.0=hadoop100:2888:3888
server.1=hadoop101:2888:3888
server.2=hadoop102:2888:3888
$ mkdir data
$ echo 0 > data/myid  # 0, 1, 2

## 配置ZooKeeper环境变量
$ vim /etc/profile.d/my.sh
export ZOOKEEPER_HOME=/opt/zookeeper
export PATH=$PATH:$ZOOKEEPER_HOME/bin
$ source /etc/profile

## 集群分发
$ xsync /opt/zookeeper
$ xsync /etc/profile.d/my.sh

集群脚本：/usr/local/bin/zk.sh

#!/bin/bash

case $1 in
"start"){
    for i in hadoop100 hadoop101 hadoop102
    do
        echo  ------------- zookeeper $i 启动 ------------
        ssh $i "/opt/zookeeper/bin/zkServer.sh start"
    done
}
;;
"stop"){
    for i in hadoop100 hadoop101 hadoop102
    do
        echo  ------------- zookeeper $i 停止 ------------
        ssh $i "/opt/zookeeper/bin/zkServer.sh stop"
    done
}
;;
"status"){
    for i in hadoop100 hadoop101 hadoop102
    do
        echo  ------------- zookeeper $i 状态 ------------
        ssh $i "/opt/zookeeper/bin/zkServer.sh status"
    done
}
;;
esac

3. Kafka 集群部署

下载地址：Apache Kafka

## 准备资源
$ cd /opt/src
$ tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/
$ cd /opt
$ mv kafka_2.12-3.0.0/ kafka
$ cd kafka

## 配置Kafka
$ vim config/server.properties
broker.id=0  # 0, 1, 2
log.dirs=/opt/kafka/data
zookeeper.connect=hadoop100:2181,hadoop101:2181,hadoop102:2181/kafka

## 配置Kafka环境变量
$ vim /etc/profile.d/my.sh
export KAFKA_HOME=/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin
$ source /etc/profile

## 集群分发
$ xsync /opt/kafka
$ xsync /etc/profile.d/my.sh

集群脚本：/usr/local/bin/kf.sh

#!/bin/bash

case $1 in
"start"){
    for i in hadoop100 hadoop101 hadoop102
    do
        echo  ------------- kafka $i 启动 ------------
        ssh $i "/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties"
    done
}
;;
"stop"){
    for i in hadoop100 hadoop101 hadoop102
    do
        echo  ------------- kafka $i 停止 ------------
        ssh $i "/opt/kafka/bin/kafka-server-stop.sh"
    done
}
;;
esac

4. 启动

# 启动ZooKeeper集群
$ zk.sh start
# 启动Kafka集群
$ kf.sh start

先停止 Kafka，等待其所有节点进程停止后，再停止 ZooKeeper：Kafka 将一些信息存储到了 ZooKeeper，停止前需要先和 ZooKeeper 通信

三、Kafka 命令行操作

1. 主题命令行操作

# 查看操作主题命令参数
$ bin/kafka-topics.sh
# 查看当前服务器中的所有topic
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092,hadoop101:9092 --list
# 创建topic（1个分区，3个副本，名为test）
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092 --create --partitions 1 --replication-factor 3 --topic test
# 查看topic详情
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092 --describe --topic test
Topic: test	TopicId: D7jWqskZSf-F-MC39chP5A	PartitionCount: 1	ReplicationFactor: 3	Configs: segment.bytes=1073741824
	Topic: test	Partition: 0	Leader: 2	Replicas: 2,1,0	Isr: 2,1,0
# 修改分区数（只能增加，不能减少）
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092 --alter --topic test --partitions 3
# 修改副本数（不能通过命令行方式修改）
# 再次查看详情
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092 --describe --topic test
Topic: test	TopicId: D7jWqskZSf-F-MC39chP5A	PartitionCount: 3	ReplicationFactor: 3	Configs: segment.bytes=1073741824
	Topic: test	Partition: 0	Leader: 2	Replicas: 2,1,0	Isr: 2,1,0
	Topic: test	Partition: 1	Leader: 0	Replicas: 0,1,2	Isr: 0,1,2
	Topic: test	Partition: 2	Leader: 1	Replicas: 1,2,0	Isr: 1,2,0
# 删除topic
$ bin/kafka-topics.sh --bootstrap-server hadoop100:9092 --delete --topic test

核心参数

参数	描述
--bootstrap-server <String: server to connect to>	连接的 Kafka Broker 的主机名称和端口号
--topic <String: topic>	操作的 topic 名称
--create	创建主题
--delete	删除主题
--alter	修改主题
--list	查看所有主题
--describe	查看主题详细描述
--partitions <Integer: # of partitions>	设置分区数
--replication-factor <Integer: replication factor>	设置分区副本
--config <String: name=value>	更新系统默认的配置

2. 生产者命令行操作

# 查看操作生产者命令参数
$ bin/kafka-console-producer.sh
# 发送消息
$ bin/kafka-console-producer.sh --bootstrap-server hadoop100:9092 --topic test
>hello

核心参数

参数	描述
--bootstrap-server <String: server to connect to>	连接的 Kafka Broker 的主机名称和端口号
--topic <String: topic>	操作的 topic 名称

3. 消费者命令行操作

# 查看操作消费者命令参数
$ bin/kafka-console-consumer.sh
# 消费topic中的数据
$ bin/kafka-console-consumer.sh --bootstrap-server hadoop100:9092 --topic test
# 把主题中所有的数据都读取出来（包括历史数据）
$ bin/kafka-console-consumer.sh --bootstrap-server hadoop100:9092 --from-beginning --topic test

核心参数

参数	描述
--bootstrap-server <String: server to connect to>	连接的 Kafka Broker 的主机名称和端口号
--topic <String: topic>	操作的 topic 名称
--from-beginning	从头开始消费
--group <String: consumer group id>	指定消费者组名称

四、其他

1. Kafka 配置文件

server.properties

# broker的全局唯一编号，不能重复，只能是数字
broker.id=0
# 处理网络请求的线程数量
num.network.threads=3
# 处理磁盘IO的线程数量
num.io.threads=8
# 发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
# 接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
# 请求套接字的缓冲区大小
socket.request.max.bytes=104857600
# kafka运行日志/数据存放的路径，无需提前创建，可以配置多个磁盘路径，用“,”分割
log.dirs=/opt/kafka/data
# topic在当前broker上的分区个数
num.partitions=1
# 恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
# 每个topic创建时的副本数
offsets.topic.replication.factor=1
# segment文件保留的最长时间，超时将被删除
log.retention.hours=168
# 每个segment文件的大小，默认最大1G
log.segment.bytes=1073741824
# 检查过期数据的时间，默认5分钟检查一次是否数据过期
log.retention.check.interval.ms=300000
# 配置连接ZooKeeper集群地址（在ZK根目录下创建“/kafka”，方便管理）
zookeeper.connect=hadoop100:2181,hadoop101:2181,hadoop102:2181/kafka

一、概述

1. Kafka 定义

2. 消息队列

(1) 传统消息队列的应用场景

(2) 消息队列的两种模式

3. Kafka 基础架构

二、 集群部署

1. 集群规划

2. ZooKeeper 集群部署

3. Kafka 集群部署

4. 启动

三、Kafka 命令行操作

1. 主题命令行操作

2. 生产者命令行操作

3. 消费者命令行操作

四、其他

1. Kafka 配置文件

二、集群部署