kafka

Apache Kafka 是一个分布式的流处理平台。它具有以下特点：

  • 支持消息的发布和订阅，类似于 RabbtMQ、ActiveMQ 等消息队列
  • 支持数据实时处理
  • 能保证消息的可靠性投递
  • 支持消息的持久化存储，并通过多副本分布式的存储方案来保证消息的容错
  • 高吞吐率，单 Broker 可以轻松处理数千个分区以及每秒百万级的消息量

架构简介

Messages and Batches

kafka基本数据单元为消息，为了提高网络使用效率，采用批写入方式

Topics and Partitions

topic为kafka消费主题，每个主题下有若干分区(partitions)，Kafka 通过分区来实现数据的冗余和伸缩性，分区可以分布在不同的服务器上。由于多个partition的特性，kafka无法保证topic范围内的消息顺序，但是可以保证单个分区内消息的顺序

broker

broker 对应着一个 kafka 的进程；一个 kafka 集群会包含多个 broker；同时需要在这些 broker中选举出一个controller，选举是通过 zk 来实现；controller 负责协调管理集群状态，同时也负责 partition 的 leader 选举；

Producers And Consumers
  • 消息的生产者，负责将消息发送到不同的 partition 中；消息的生产需要考虑幂等性、正确性以及安全性；kafka 引入了 ack 机制；ack 为 0，则不需要 kafka 回复，此时可能造成数据丢失；ack为 1， 则需要等待 leader 回复，此时其他 replica 可能还没同步 leader 挂掉，数据安全性没法得到保证；ack 为 -1，则需要等待其他 replica 同步完成后，才回复，此时数据最健壮，但是效率最低；
  • 消息的消费者，负责消费消息；一个 partition 对应一个consumer， 而一个 consumer 可以对应多个 partition；消费同一类消息的高吞吐量，可以设置 consumer group；
副本同步策略

每个分区里有多个副本，这些副本有一个leader。只有副本全部同步完成才发送ack。这里指同步策略，是全量同步，而不是半数以上同步了就认为该数据已经commit。不过也可以设置最少同步副本数提高性能（min.insync.replicas）

ISR

Leader 维护了一个动态的 in-sync replica set (ISR)，意为和 leader 保持同步的 follower 集合。当 ISR 中的 follower 完成数据的同步之后，leader 就会给 producer 发送 ack。如果 follower 长时间未向 leader 同步数据，则该 follower 将被踢出 ISR，该时间阈值由 replica.lag.time.max.ms 参数设定。Leader 发生故障之后，就会从 ISR 中选举新的 leader。

数据可见性

需要注意的是，并不是所有保存在分区首领上的数据都可以被客户端读取到，为了保证数据一致性，只有被所有同步副本 (ISR 中所有副本) 都保存了的数据才能被客户端读取到。

kafka读写机制

producer写流程

producer写入消息流程如下：

  • 连接ZK集群，从ZK中拿到对应topic的partition信息和partition的Leader的相关信息

  • 连接到对应Leader对应的broker

  • 将消息按批次发送到partition的Leader上

  • 其他Follower从Leader上复制数据

  • 依次返回ACK

  • 直到所有ISR中的数据写完成，才完成提交，整个写过程结束

consumer 读流程

  • 连接ZK集群，从ZK中拿到对应topic的partition信息和partition的Leader的相关信息

  • 连接到对应Leader对应的broker

  • consumer将自己保存的offset发送给Leader

  • Leader根据offset等信息定位到segment（索引文件和日志文件）

  • 根据索引文件中的内容，定位到日志文件中该偏移量对应的开始位置读取相应长度的数据并返回给consumer

kafka集群选举

副本leader选举

只有完全追上Leader数据的follower才能进行选举，Leader发生故障之后，会从ISR中选出一个新的Leader

controller选举

这部分由ZK完成，不过高本版kafka引入kratf，就可以完成去ZK化了。 ratf是一种简单易理解并且严格复合数学归纳的共识算法。

自测环境搭建

zk

docker pull wurstmeister/zookeeperdocker run -itd --name zookeeper -p 2181:2181 wurstmeister/zookeeper

kafka

 docker pull wurstmeister/kafka docker run -itd --name kafka -p 9092:9092 -e HOST_IP=10.74.18.61 -e KAFKA_ADVERTISED_PORT=9092 -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 -e KAFKA_ADVERTISED_HOST_NAME=10.74.18.61 --link zookeeper:zookeeper wurstmeister/kafka

go链接kafka生产消费

go版本：1.21
生产者

package mainimport ("fmt""github.com/IBM/sarama")func main() {config := sarama.NewConfig()// 等待服务器所有副本都保存成功后的响应，对应ack=-1config.Producer.RequiredAcks = sarama.WaitForAll// 随机的分区类型：返回一个分区器，该分区器每次选择一个随机分区config.Producer.Partitioner = sarama.NewRandomPartitioner// 是否等待成功和失败后的响应config.Producer.Return.Successes = true// 使用给定代理地址和配置创建一个同步生产者producer, err := sarama.NewSyncProducer([]string{"10.74.18.61:9092"}, config)if err != nil {panic(err)}defer producer.Close()//构建发送的消息，msg := &sarama.ProducerMessage{//Topic: "test",//包含了消息的主题Partition: int32(10), //Key: sarama.StringEncoder("key"), //}var value stringvar msgType stringfor {_, err := fmt.Scanf("%s", &value)if err != nil {break}fmt.Scanf("%s", &msgType)fmt.Println("msgType = ", msgType, ",value = ", value)msg.Topic = msgType//将字符串转换为字节数组msg.Value = sarama.ByteEncoder(value)//fmt.Println(value)//SendMessage：该方法是生产者生产给定的消息//生产成功的时候返回该消息的分区和所在的偏移量//生产失败的时候返回errorpartition, offset, err := producer.SendMessage(msg)if err != nil {fmt.Println("Send message Fail", err)}fmt.Printf("Partition = %d, offset=%d\n", partition, offset)}}

消费者

package mainimport ("fmt""sync""github.com/IBM/sarama")var (wg sync.WaitGroup)func main() {// 根据给定的代理地址和配置创建一个消费者consumer, err := sarama.NewConsumer([]string{"10.74.18.61:9092"}, nil)if err != nil {panic(err)}//Partitions(topic):该方法返回了该topic的所有分区idpartitionList, err := consumer.Partitions("test")if err != nil {panic(err)}for partition := range partitionList {//ConsumePartition方法根据主题，分区和给定的偏移量创建创建了相应的分区消费者//如果该分区消费者已经消费了该信息将会返回error//sarama.OffsetNewest:表明了为最新消息pc, err := consumer.ConsumePartition("test", int32(partition), sarama.OffsetNewest)if err != nil {panic(err)}defer pc.AsyncClose()wg.Add(1)go func(sarama.PartitionConsumer) {defer wg.Done()//Messages()该方法返回一个消费消息类型的只读通道，由代理产生for msg := range pc.Messages() {fmt.Printf("%s---Partition:%d, Offset:%d, Key:%s, Value:%s\n", msg.Topic, msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))}}(pc)}wg.Wait()consumer.Close()}