Kafka是最前沿的开源MQ之一,阿里的RocketMQ也借鉴了不少Kafka的思想。2011年领英发了篇文章描述Kafka的设计,我这先学习初版。
新版最重要的改变就是exactly once,众所周知,at least once很容易,retry即可; 而exactly once则很难, 它必须同时维护幂等性。
Reference: http://notes.stephenholiday.com/Kafka.pdf
经典的生产者消费者模型。一个broker可能持有多个topic,每个topic又可能有多个partition。
还有一个东西没有画出来,就是zookeeper,管理metadata。
//Producer
producer = new Producer(…);
message = new Message(“test message str”.getBytes());
set = new MessageSet(message);
producer.send(“topic1”, set);
//Consumer
streams[] = Consumer.createMessageStreams(“topic1”, 1)
for (message : streams[0]) {
bytes = message.payload();
// do something with the bytes
}
简易存储
事实上Kafka原本是作为日志系统,供其offline处理的。
append log
每个partition对应连续的逻辑log,由固定大小的一组物理log segment组成。log按照pub数目或者时间周期性地进行flush持久化。
log直接按照逻辑offset映射,而不是通过显式id索引,这里的id并不是连续的,而是通过上一条的id增加msg长度计算的。
Broker在内存中维护偏移量的有序链表,包括每个segment第一个msg的offset
(这里不知道是不是把这些offset给单独存储了,一方面充当跳表容易定位,一方面容易计算出在segment中的物理offset)
效率传输
无状态
Kafka本身并不记忆什么数据被消费,而是只删除过期(自定义)的log。如果是push model,那么显然需要kafka记忆;而pull model则只需要consumer记忆。Consumer因此能随时消费之前的消息,例如假如consumer需要持久化消息,他可以做checkpoint,然后恢复的时候redo log。
Producer可以随机或者按照partition函数映射到对应的broker。
一组Consumer称为Consumer Group,合起来消费某个topic,每个msg只被一个Consumer消费;而Group之间则独立。这里其实就相当于负载均衡,因此要避免重复消费。
(后面也可以broadcast)
partition是并行基本单元,每次仅有一个group内的consumer能消费。
这里利用zookeeper实现
For each topic T that Ci subscribes to {
remove partitions owned by Ci from the ownership registry
read the broker and the consumer registries from Zookeeper
compute PT = partitions available in all brokers under topic T
compute CT = all consumers in G that subscribe to topic T
sort PT and CT
let j be the index position of Ci in CT and let N = |PT|/|CT|
assign partitions from j*N to (j+1)*N - 1 in PT to consumer Ci
for each assigned partition p {
set the owner of p to Ci in the ownership registry
let Op = the offset of partition p stored in the offset registry
invoke a thread to pull data in partition p from offset Op
}
}
但这样因为负载均衡都是本地进行的,consumer彼此不通信。有的consumer会尝试pull那些仍然属于其他consumer的partition,这种情况它会释放自己的消费的partition然后等待一会儿之后retry rebalance
新增的consumer group的offset可能是log offset的最小或者最大值,根据配置而定。
初版的Kafka仅仅保证At least once,因为领英暂时不需要exactly once。现在的exactly once是在producer增加了id用于去重,同时提供了分布式事务支持
同时Kafka仅仅能保证单个partition有序(append log),而无法保证topic有序
Kafka在log加入CRC(循环冗余校验)避免log污染
初版的Kafka没有备份机制,现在的kafka是主从备份,平时只有leader服务。
Kafka本身可以作为其他Kafka的producer和consumer
因为Kafka只支持无类型字节流,使用Avro作为序列化协议,在里面存储了schema ID提供类型信息,然后再反序列化。