1. Kafka基础架构

  • Producer
    • 消息生产者,负责将数据推送给 broker 的 topic
  • Consumer
    • 消息消费者,负责从 broker 的 topic 中拉取数据,并自己进行处理
  • Consumer Group
    • consumer group 是 kafka 提供的可扩展且具有容错性的消费者机制
    • 一个消费者组可以包含多个消费者
    • 一个消费者组有一个唯一的 ID(group Id)
    • 消费者组内每个消费者负责消费不同分区的数据,一个分区只能由一个组内消费者消费
    • 所有的消费者都属于某个消费者组,即消费者组是逻辑上的一个订阅者
    • 消费者组之间互不影响
  • Broker
    • 一台 kafka 服务器就是一个 broker,一个 broker 可以容纳多个topic
    • 一个集群由多个 broker 组成,这样才能实现负载均衡、以及容错
    • broker是无状态(Sateless)的,它们是通过 ZooKeeper 来维护集群状态
    • 一个Kafka的broker每秒可以处理数十万次读写,每个broker都可以处理TB消息而不影响性能
  • Zookeeper
    • ZK 用来管理和协调 broker,并且存储了 Kafka 的元数据(例如:有多少topic、partition、consumer)
    • ZK 服务主要用于通知生产者和消费者 Kafka 集群中有新的 broker 加入、或者 Kafka 集群中出现故障的 broker
    • Kafka 正在逐步想办法将 ZooKeeper 剥离,维护两套集群成本较高,社区提出 KIP-500 就是要替换掉 ZooKeeper 的依赖。“Kafka on Kafka”——Kafka 自己来管理自己的元数据
  • Partitions
    • 为了实现扩展性,一个非常大的 topic 可以分布到多个broker(即服务器)上,一个 topic 可以分为多个 partition,每个 partition 是一个有序的队列
  • Replicas
    • 副本可以确保某个服务器出现故障时,确保数据依然可用
    • 一个 topic 的每个分区都有若干个副本,一个 leader 和若干个 follower
    • 在Kafka中,一般都会设计副本的个数>1
  • Topic
    • 主题是一个逻辑概念,用于生产者发布数据,消费者拉取数据
    • Kafka 中的主题必须要有标识符,而且是唯一的,Kafka 中可以有任意数量的主题,没有数量上的限制
    • 在主题中的消息是有结构的,一般一个主题包含某一类消息
    • 一旦生产者发送消息到主题中,这些消息就不能被更新(更改)
  • Offset
    • offset 记录着下一条将要发送给 Consumer 的消息的序号
    • 默认 Kafka 将 offset 存储在 ZooKeeper 中
    • 在一个分区中,消息是有顺序的方式存储着,每个在分区的消费都是有一个递增的 id,这个就是偏移量 offset
    • 偏移量在分区中才是有意义的。在分区之间,offset是没有任何意义的
  • leader
    • 每个分区多个副本的“主”,生产者发送数据的对象,以及消费者消费数据的对象都是 leader
  • follower
    • 每个分区多个副本中的“从”,实时从 leader 中同步数据,保持和 leader 数据的同步。leader 发生故障时,某个 follower 会成为新的 follower。

2. Kafka的命令行操作

  • 启动内置zookeeper
1
2
./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
-daemon 使用守护进程启动
  • 启动kafka
1
2
./bin/kafka-server-start.sh -daemon config/server.properties
-daemon 使用守护进程启动
  • 查看当前服务器中的所有topic
1
./bin/kafka-topics.sh --zookeeper IP:2181 --list
  • 创建topic
1
2
3
4
./bin/kafka-topics.sh --zookeeper IP:2181 --create --replication-factor 1 --partitions 1 --topic test
--topic 定义topic名
--replication-factor 定义副本数
--partitions 定义分区数
  • 删除topic
    • 需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启
1
./bin/kafka-topics.sh --zookeeper IP:2181 --delete --topic test
  • 生产者发送消息
1
./bin/kafka-console-producer.sh --broker-list IP:9092 --topic test
  • 消费者消费消息
1
2
3
./bin/kafka-console-consumer.sh --bootstrap-server IP:9092 --from-beginning --topic test --consumer-property group.id=group_1
--from-beginning 会把主题中以往所有的数据都读取出来,根据业务场景选择是否增加该配置
--consumer-property group.id=group_1 指定消费者组为group_1
  • 查看某个Topic的详情状态和分区负载详情
1
./bin/kafka-topics.sh --zookeeper IP:2181 --describe --topic test
  • 查看某一个topic最大最小的offset
1
2
3
4
# 查看最小offset
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list IP:9092 --topic test --time -2
# 查看最大offset
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list IP:9092 --topic test --time -1
  • 查看所有消费者组
1
./bin/kafka-consumer-groups.sh --bootstrap-server IP:9092 --list
  • 查看某一消费者组的详细信息
1
./bin/kafka-consumer-groups.sh --bootstrap-server IP:9092 --group group_1 --describe
  • 删除某一消费者组
1
./bin/kafka-consumer-groups.sh --bootstrap-server IP:9092 --delete --group group_1