本文将介绍最常用的分布式消息中间件kafka。由于笔者水平受限，因此介绍不一定全面，也不会太深入，仅供参考。

环境

• windows10
• kafka 2.1.1

安装

• kafka依赖于zookeeper，因此需要先安装zookeeper
• windows环境下kafka只需解压即可使用，但是需要注意解压路径不能包括空格
• 执行kafka-server-start.bat config\server.properties启动服务

提示命令语法不正确

如果启动时提示命令语法不正确，那么需要在kafka安装目录中找到bin\windows目录中的kafka-run-class.bat，为set COMMAND后面的%CLASSPATH%加上双引号

基本概念

• topic，一个topic可以理解为一个逻辑上的消息队列
• partition，一个topic包括一个或多个partition，同一个partition的数据必定在同一台机器上
• segment，partition以segment存储，每个segment大小相同，对应物理存储文件
• retention period，消息保留时间，超过被丢弃
• consume group，消费组，kafka对同一个消费组内的实例做负载均衡，对不同消费组做广播。一个partition会被指派到消费组里的某个实例。
• leader和followers，partition的主备模型

命令行操作

主题操作

• 创建主题

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

• 查看主题

kafka-topics.bat --list --zookeeper localhost:2181

注意2.2版本可以直接用--bootstrap-server替代--zookeeper

生产消费操作

• 生产消息

kafka-console-producer.bat --broker-list localhost:9092 --topic test

• 消费消息

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning

消费组操作

• 查看consumer group的offset

kafka-consumer-groups.bat --bootstrap-server localhost:9092 --describe --group my-group

• 查看所有的consumer group

kafka-consumer-groups.bat --bootstrap-server localhost:9092 --list

• 查看consumer group的所有成员

kafka-consumer-groups.bat --bootstrap-server localhost:9092 --describe --group my-group --members

• 删除consumer group

kafka-consumer-groups.bat --bootstrap-server localhost:9092 --delete --group my-group --group my-other-group

• 修改consumer group的offset

kafka-consumer-groups.bat --bootstrap-server localhost:9092 --topic topic_name --group group_name --execute --reset-offsets --to-offset 0

生产者/消费者

kafka消息格式

一条消息是一个record batch，包含record batch header，每条record又有各自的header

message delivery semantics，消息投递语义

• at most once，最多一次，消息可能会丢失，但不会重复
  先commit offset，再进行业务处理
• at least once，最少一次，消息不会丢失，可能会重复
  先进行业务处理，再commit offset
• exactly once，只消费一次

如何实现exactly once

• 生产者：0.11.0.0之后的版本实现了幂等操作
• 消费者：或者实现幂等，或者把commit offset和业务处理绑定成一个事务
• 使用kafka streams可以实现

leader

kafka文件存储格式

一个segment由.index和.log组成。.index是索引文件，记录每条消息的offset和在.log中的地址，.log中存储具体的数据。segment大小固定，但是包含不同数目的消息，segment文件的命名由上一个segment的最后一条消息的offset决定。查询指定offset消息的过程是先通过二分查找找到对应的segment，然后在.index文件中通过二分查找找到对应的存储地址。

kafka高性能实现

• kafka利用内存映射文件技术，包括mmap和sendfile
• kafka客户端会缓存数据然后批量操作
• byte copy，从生产者到kafka到消费者都是二进制数据，不需要转换

log compaction

compaction指对相同key的数据进行合并。

• 开启log compaction

log.cleanup.policy=compact

• 开启compaction后，默认情况下除了最后一个segment，都会进行合并。

followers

kafka的高可用设计

每个partition都有一个leader，若干个followers，读写请求发送给leader处理。leader维护了一个isr（in-sync replicas）列表，写数据时只有当指定数量的isr告知已收到（acknowledge）leader才会commit，而数据只有commit之后才会被消费者看到。告知已收到的数量可以由producer决定，包括0,1或者all（-1）

崩溃恢复机制

如果分区的当前leader挂掉了，会从isr列表中重新选举leader。如果列表中的所有节点都挂掉了，那么有以下几种策略

• 等待isr列表中的节点恢复，然后成为leader
• 等待第一个节点恢复（可能不在isr列表中），然后成为leader

kafka vs rocketmq

• 定位：kafka主要定位于日志输出，而rocketmq定位于非日志的可靠消息传输
• 存储：kafka多文件存储，rocketmq单文件存储，有人说kafka吞吐量更高，也有人说rocketmq支持更多的topic（性能下降平缓），不过据说kafka 1.1.0之后可以支持20万的partition
• 备份：rocketmq没有master选举；rocketmq的slave可读，而kafka只能从leader读