基本介绍
![]()
- 最初由Linkedin公司开发，2010捐给Apache
- 特性：分布式、分区、多副本
- 基于zk协调的分布式消息系统
- 可实时处理大数据量消息
- 应用场景：基于hadoop的批处理、Spark流失处理、web服务器日志、系统访问日志等
- scala语言开发
业务应用场景
![]()
基本概念
- Broker：消息处理节点，可理解为就是一个kafka节点
- Topic：对消息进行归类
- Producer/Consumer：生产者与消费者
- ConsumerGroup：消费组，一条消息可被多个不同消费组消费，同一消费组内，只有一个消费者能消费某条消息
- Partition：分区，物理概念，一个Topic可分为多个分区，每个分区内的消息是有序的；个人理解与RocketMQ中的Queue概念相似
![]()
基本使用
- 环境准备：kafka为Scala语言开发，需运行在jvm上，所以需准备JDK环境
- kafka依赖zk做协调器，需安装zk
- 下载kafka，主要修改config/server.properties
- 关键属性：
#broker.id属性在kafka集群中必须要是唯一
broker.id=0
#kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://192.168.65.60:9092
#kafka的消息存储文件
log.dir=/usr/local/data/kafka-logs
#kafka连接zookeeper的地址，ZK集群以逗号隔开
zookeeper.connect=192.168.65.60:2181
- 启动脚本语法：kafka-server-start.sh [-daemon] server.properties
- 启动后可进入zk查看zk下的节点数据
- 停止：bin/kafka-server-stop.sh
创建主题
- 主题默认：分区数为1，副本数为1，最小写成功数为1
bin/kafka-topics.sh --create --zookeeper 192.168.65.60:2181 --replication-factor 1 --partitions 1 --topic test
发送消息
- 自带
bin/kafka-console-producer.sh命令客户端
bin/kafka-console-producer.sh --broker-list 192.168.65.60:9092 --topic test
>this is a msg
>this is a another msg
消费消息
- 自带消费消息命令终端：
bin/kafka-console-consumer.sh
bin/kafka-console-consumer.sh --bootstrap-server 192.168.65.60:9092 --topic test
- 带参数
--from-beginning，表示可消费之前全部消息
- 带参数
--whitelist "test|test-2" 同时消费多个主题
单播与多播
- 单播：一条消息只让一个消费者消费，类似Queue模式；我们让所有消费者在同一个消费者内即可实现；
- 多播：一条消息可让多个消费者同时都消费，类似发布订阅模式；我们让每个消费者存在不同消费组即可实现；
其他命令
// 查看消费组组名
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.65.60:9092 --list
// 查看消费组消费偏移量
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.65.60:9092 --describe --group testGroup
current-offset：当前消费组的已消费偏移量
log-end-offset：主题对应分区消息的结束偏移量(HW)
lag：当前消费组未消费的消息数
![]()
Topic与Log
- Topic即为一个消息类别名称，同类消息一般发到同一Topic下
- 每个Topic下对应有多个分区日志文件
![]()
- 分区是一个有序的消息队列，消息被按序添加到commit log文件中
- 每个消息在一个分区中有唯一的标识，称为offset
- 不同分区中的消息，其offset可能相同
- 消息一般不会自动删除，默认被保留约最近7天
- 消费者基于自己在提交日志中消费的进度（offset）进行工作
- offset由消费者自己维护
- 所以消费者可按序逐条消费commit log消息，也可指定offset来重复或跳过消费某些消息
// 创建2个分区的主题
bin/kafka-topics.sh --create --zookeeper 192.168.65.60:2181 --replication-factor 1 --partitions 2 --topic test1
查看下topic的情况
bin/kafka-topics.sh --describe --zookeeper 192.168.65.60:2181 --topic test1
![]()
- leader节点：负责给定分区的所有读写请求
- replicas节点：某分区存在的备份，不管是不是leader，甚至中途挂了都会列出
- isr：是replicas的子集，只列存活且一同步备份了该分区的节点
- 消息日志主要存在分区文件夹以log结尾的文件
- 目前支持对topic分区数据量进行动态扩容，单不支持缩减
思考：为什么需要分区？ 日志文件受到机器文件大小限制，分区后则可将不同分区放到集群中的不同机器上，相当于分布式存储；其次为了提高生产消息与消费消息的并行度；
