Kafka Agent 设计 - 可靠事件记录不是一件简单的事情

用 golang 编写，解决事件快速同时可靠入Kafka的问题。绿色表示goroutine（多是不一样的线程），蓝色表示共享的资源。全部对蓝色资源的访问须要加锁。磁盘队列使用内存映射文件实现ring buffer。kafka agent启动的时候有一个bootstrap的broker列表，同时发送的时候会与每一个相关的broker维护一个到多个链接。可是kafka agent不读取也不监听zookeeper。

收益

• 业务方不须要等待kafka响应就能够继续干别的去（低延迟）

• 基于磁盘的队列（高可靠）

• 消息按partition排队后组大包并压缩发送（高吞吐）

• 由于发送到kafka以前业务侧已经拿到响应并返回，kafka的request/response能够简单地一去一回，不用基于corelation id回调通知业务侧（简单设计）

缺点是业务侧没法知道写入的kafka offset。

生产过程

业务方和 Kafka Agent 能够在同一个进程内，也能够经过本地unix domain socket沟通。若是是在同进程内，事件直接在业务方的 goroutine 内产生并入磁盘队列。若是是在不一样进程内，事件经过本地网络请求转发，由kafka agent代为入队列。入队列成功以后有返回，业务侧接到返回则认为事件已经可靠保存，Eventually会进入到Kafka中。

1. 产生事件，生成hash值。查询 queue registry 得到topic的partition数量。根据hash值计算落到parition号。根据topic和partition找到对应的磁盘队列。

2. 若是没有对应的磁盘队列，则新建，并保存回 queue registry。若是没有对应 topic 的 metadata 则须要查询并保存回 queue registry。

3. 事件写入磁盘队列，移动写入offset的指向位置。若是写入超过读取速度，则覆盖（丢弃掉旧的event）。

4. 返回业务侧，告知已经可靠写入

异常流程

• 没有topic的metadata

• 没有对应的queue

• topic的partition数量可能增长，须要定时刷新

• 写入会超过读取的速度

• 写入超过读取速度的时候，读取可能正在进行中

发送过程

每一个队列有一个对应的goroutine负责发送到kafka。

1. 定时唤醒发送goroutine，从本身负责的queue里读取一批事件（只会对应一个topic的一个partition，由于queue是对应topic和partition建立的）。

2. 当前goroutine里若是没有缓存对应的broker（parition的leader）的connection pool，则从全局链接池里拿一个pool。从pool里借一个conn。

3. 若是pool不存在则新建pool。

4. 发送消息到kafka broker，并同步等待其返回。而后归还connection到pool里。

5. 此时消息已经可靠写入kafka，移动读取的offset。

异常流程

• partition对应的broker还未知

• broker对应的pool还未建立

• 从broker里取得的connection已经损坏

• 写入时broker告知leader已经改变

• 移动offset的时候写入已经溢出

Reliable Event Logging

不少时候咱们都但愿可靠地记录事件，这些事件处理大体分为两类

日志监控

• 记录监控指标，用于告警

• 记录原始日志，用于定位问题

• 记录原始事件，用于离线统计分析

业务事件

• 离线计算发放奖励

• 触发一系列周边系统的动做

日志监控类的要求是迅速量大，能够丢个别的日志。业务事件类的要求是一条也不能丢，不然业务方要引入另一个队列作离线对帐（这队列不仍是logging么，通常是业务本身的主数据库来承担）。业务事件的模式不能依靠上游来重试，由于主业务流程已经完成了，事件的写入和接下来的处理不该该来决定业务操做自己的成败。简单来讲就是有一些业务但愿本身业务自己成功了，事件必须可靠记录，同时基于事件的后续处理必须发生。

如今对kafka的使用方式，主要有两种

• kafka producer => kafka，实现方式是内存作buffer，而后批量写入

• log file => 日志采集 => kafka 用日志文件作缓冲

两种方式都不够完美，都没法知足使用方的要求。日志方式的问题是使用麻烦，并且性能开销大。直接用producer的方式几乎无持久化作缓冲，大部分时候就是靠内存buffer活着（除非可以忍受直接写远程kafka的延迟），根本没有持久化保证可靠性。

因此事实上目前基于kafka的应用架构里都没法把kafka作为一个可靠存储来用，不是由于kafka自己不可靠，而是作到不拖慢业务的同时可靠入库并不容易。或者讲究着把一个不可靠的data pipeline看成可靠的来用。

More on this: http://www.slideshare.net/JiangjieQin/no-data-loss-pipeline-with-apache-kafka-49753844