Kafka

一、Kafka是一个分布式消息队列系统，以集群方式存在。

优点：高吞吐率（每秒百万级）；基于多分区多副本实现高容错，并发能力强；易扩展（增加blocker）等。

缺点：仅保证分区内有序，不保证topic全局消息有序。

二、内部原理及实现

Kafka依赖zk，以集群方式工作，每台机器称为一个blocker（与其他分布式环境不同的是，仅有1台blocker也能工作），并分别指定blockerId，kafka中同一类型数据以topic形式存在，可对topic进行分区，及指定副本数，例如可将 topic1 分区为3个partition:p0,p1,p2 ;每个分区副本数为3，则每个分区块会均匀的分布在blocker中，对于每个partition，有多个副本，会选出一个leader partition对外接收请求（只有leader partition会接收外部请求，其他follower partition只负责同步leader数据，不接收外部请求，包括读请求，这里与zk不同，zk的follower会接收读请求）。集群有多个blocker，会基于zk选择出一个Controller，来管理整个集群，包括blocker管理、topic管理，partition leader分配等。

每个blocker只存储消息体，不存储元数据（topic信息；分区信息：topic有哪些分区，哪些副本，分别在哪台blocker上，哪个是leader；consumer信息及读取消息后提交的偏移量数据等），元数据存储在zk中。

Controller主要功能：

（1）更新元数据

若blocker宕机，相关副本不可用，需要重新选择相关partition的leader，及更新ISR列表，选举完毕后，controller会更新zk中的元数据，并广播给其他blocker（每个blocker都可以对外提供元数据信息）

（2）topic管理

新增、删除topic

（3）分区重新分配

通过自带的 kafka-reassign-partitions 脚本可重新制定topic分区，由controller处理请求，重新分配并选择每个分区的leader。

（4）partition leader 分配

partition  leader 选举的原理 ：

　　1. 在ZooKeeper上创建/admin/preferred_replica_election节点，并存入需要调整Preferred Replica的Partition信息。

　　2. Controller一直Watch该节点，一旦该节点被创建，Controller会收到通知，并获取该内容。

　　3. Controller读取Preferred Replica，如果发现该Replica当前并非是Leader并且它在该Partition的ISR中，

Controller向该Replica发送LeaderAndIsrRequest，使该Replica成为Leader。

如果该Replica当前并非是Leader，且不在ISR中，Controller为了保证没有数据丢失，并不会将其设置为Leader。

（总结：副本partition向zk创建临时节点并写入数据，Controller broker监听节点，一旦创建成功，获取其中的消息，更新副本Leader及ISR列表，完成选举；只会选择ISR列表中的设备作为leader）

具体实现：

在kafka中创建一个topic，controller会接收创建的请求，并在zk的目录下新增topic节点

zk中会维护consumer相关信息：

其他还有什么信息查查zk节点就知道了。。。

三、消息生产与消费的流程

Producer 发送消息到kafka中：

生产者会指定一个blocker servers，与集群中blocker一致。并指定acks级别（发送完消息后是否等待消息成功写入kafka的反馈）。

kafka集群中的任何一个broker,都可以向producer提供metadata信息,这些metadata中包含”集群中存活的servers列表”/”partitions leader列表”等信息(请参看zookeeper中的节点信息).

当producer获取到metadata信息之后, producer将会和Topic下所有partition leader保持socket连接;

消息由producer直接通过socket发送到broker,中间不会经过任何”路由层”.事实上,消息被路由到哪个partition上,由producer客户端决定.比如可以采用”random””key-hash””轮询”等,如果一个topic中有多个partitions,那么在producer端实现”消息均衡分发”是必要的.

在producer端的配置文件中,开发者可以指定partition路由的方式.

Consumer 消费：

需要为每个Consumer 指定一个Consumer Group，不同Group可读取相同消息；同一group内不同consumer 互斥，不会消费相同的消息。

consumer 相关信息也在zk中维护（属于哪个consumer group，读取哪个topic，不同的分区分别读到哪里的偏移量等）；consumer读取完消息后可选择自动提交偏移量 或 手动提交；

Consumer均衡算法：

当一个group中,有consumer加入或者离开时,会触发partitions均衡.均衡的最终目的,是提升topic的并发消费能力.

1) 假如topic1,具有如下partitions: P0,P1,P2,P3

2) 加入group中,有如下consumer: C0,C1

3) 首先根据partition索引号对partitions排序: P0,P1,P2,P3

4) 根据consumer.id排序: C0,C1

5) 计算倍数: M = [P0,P1,P2,P3].size / [C0,C1].size,本例值M=2(向上取整)

6) 然后依次分配partitions: C0 = [P0,P1],C1=[P2,P3],即Ci = [P(i * M),P((i + 1) * M -1)]

一般情况下可以 1个 consumner负责一个 partition，如果partition 比consumer多，可以1个consumer负责多个partition  提升并发能力。。

  1) Producer端直接连接broker.list列表,从列表中返回TopicMetadataResponse,该Metadata包含Topic下每个partition leader建立socket连接并发送消息.

    2) Broker端使用zookeeper用来注册broker信息,以及监控partition leader存活性.

    3) Consumer端使用zookeeper用来注册consumer信息,其中包括consumer消费的partition列表等,同时也用来发现broker列表,并和partition leader建立socket连接,并获取消息.

Flume

一、Flume是海量数据采集、聚合、传输的系统，广泛用于日志采集。

内置丰富的组件，通过组件间的自由组合，实现不同架构的数据采集系统，无需开发，只需手动配置文件即可实现。

理解下图：