一、协议设计

kafka自定义了一组基于TCP的二进制协议，只要遵守这组协议的格式，就可以向kafka发送消息或者拉取消息等，下面是kafka发送消息请求和返回体格式：

消息发送请求体格式

• api_key：API标识，比如PRODUCE、FETCH分别表示发送消息和拉取消息的请求
• api_version：API版本号
• correlation_id：本次请求的唯一id，用来和返回体对应
• client_id：客户端Id

消息发送返回体格式

1、消息发送

请求体

消息发送/消息返回的api_key = 0，表示PRODUCE，而目前最新版本为V6，那么api_version = 6

• transactional_id：事务id
• acks: ack机制
• timeout：超时时间
• topic_data：根据topic来分类的数据集合
  • topic：主题
  • data：某个主题下的数据
    • partition：分区
    • record_set：分区下的数据

返回体

• throttle_time_ms：超过配额限制则需要延迟该请求的处理时间
• responses：返回的数据集合，按照主题分区的粒度进行区分
  • topic：主题
  • partition_responses：分区返回信息
    • partition：分区
    • erro_code：错误码
    • base_offset：消息集的初始偏移量
    • log_append_time：消息写入broker端的时间
    • log_start_offset：所在分区的起始偏移量

2、消息拉取

请求体

对应的api_key=1，表示FETCH，请求格式如下：

• replica_id：用来指定副本的brokerId
• max_wait_time
• min_bytes
• max_bytes
• isolation_level
• session_id
• epoch
• topics：所要拉取的主题信息
  • topic
  • partitions
    • partition：分区编号
    • fetch_offset：从分区的那个位置开始读取消息
    • log_start_offset：分区的起始偏移量
    • max_bytes
• forgotten_topic_s_data

返回体

二、时间轮

kafka中大量的延时操作，比如延时生产、延时拉取和延时删除等，kafka是基于时间轮的概念自定义实现了一个用于延时功能的定时器

数组中每个元素可以存放一个定时任务列表，这个定时任务列表是一个环形的双向链表，时间轮是由多个时间格组成的，每个时间格代表当前时间轮的基本时间跨度（tickMs），时间格的数量（wheelSize）是固定的，那么一个时间片轮转的总时间就是tickMs * wheelSize。

假设现在某个格子上已经存在一个延时任务，如果另一个大延迟任务也能落在这个格子上，如果使用链表的形式去堆积，这个大延迟任务迟迟不进行，会严重影响kafka性能，所以kakfa引入了层级时间轮。

常见的钟表就是这种类似三层结构的时间轮，第一层时间轮为秒钟，第二层时间轮为分钟，第三层时间轮为小时，相当于存在一个时间轮升级和时间轮降级的操作。

三、延迟操作

kafka中有多种延迟操作，比如延时生产、延时拉取和延时数据删除等。延时操作需要延迟返回响应的结果，它必须有一个超时时间，如果没有在这个超时时间内没有完成既定的任务，那么就需要强制完成以返回响应结果给客户端。

对于延时生产操作而言，它的外部事件是所要写入消息的某个分区的HW发生增长。

延时操作创建之后会被加入延时操作管理器来做专门的处理，延时操作可能会超时，每个延时操作管理器会配备一个定时器来做超时管理。

参考

官方文档 (opens new window)

图解Kafka之实战指南 (opens new window)