一、认识kafka

Kafka到底是个啥？用来干嘛的？

官方定义如下：
Kafka is used for building real-time data pipelines and streaming apps. It is horizontally scalable, fault-tolerant, wicked fast, and runs in production in thousands of companies.
大致的意思就是，这是一个实时数据处理系统，可以横向扩展，并高可靠！

实时数据处理：从名字上看，就是将数据进行实时处理，在现在流行的微服务开发中，最常用实时数据处理平台有 RabbitMQ、RocketMQ 等消息中间件。
这些中间件，最大的特点主要有两个：

• 服务解耦
• 流量削峰
  在早期的 web 应用程序开发中，当请求量突然上来了时候，我们会将要处理的数据推送到一个队列通道中，然后另起一个线程来不断轮询拉取队列中的数据，从而加快程序的运行效率。
  
  但是随着请求量不断的增大，并且队列通道的数据一致处于高负载，在这种情况下，应用程序的内存占用率会非常高，稍有不慎，会出现内存不足，造成程序内存溢出，从而导致服务不可用。
  随着业务量的不断扩张，在一个应用程序内，使用这种模式已然无法满足需求，因此之后，就诞生了各种消息中间件，例如 ActiveMQ、RabbitMQ、RocketMQ等中间件。
  采用这种模型，本质就是将要推送的数据，不再存放在当前应用程序的内存中，而是将数据存放到另一个专门负责数据处理的应用程序中，从而实现服务解耦。
  
  消息中间件：主要的职责就是保证能接收到消息，并将消息存储到磁盘，即使其他服务都挂了，数据也不会丢失，同时还可以对数据消费情况做好监控工作。

应用程序：只需要将消息推送到消息中间件，然后启用一个线程来不断从消息中间件中拉取数据，进行消费确认即可！

Kafka 本质其实也是消息中间件的一种，最初为了解决数据管道问题，LinkedIn团队采用了 ActiveMQ 来进行数据交换，在2010年左右，那时ActiveMQ 还远远无法满足 LinkedIn团队 对数据传递系统的要求，经常由于各种缺陷而导致消息阻塞或者服务无法正常访问，为了能够解决这个问题，LinkedIn 决定研发自己的消息传递系统，Kafka 由此诞生。

Kafka特性：

• 高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作。
• 可扩展性：集群支持热扩展（kafka-reassign-partitions.sh）分区重分配、迁移
• 持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失
• 容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）
• 高并发：支持数千个客户端同时读写

Kafka使用场景

• 日志收集：可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。（ELK+kafka）
• 消息系统：解耦和生产者和消费者、缓存消息等。
• 用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和数据挖掘。
• 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。
• 流式处理：比如spark streaming和storm（大数据流处理框架，Storm是纯实时处理数据,SparkStreaming是微批处理数据）
• 事件源 可以使用kafka流来处理事件

二、架构介绍

角色术语：

• Producer：Producer 即生产者，消息的产生者，是消息的入口
• Broker：Broker 是 kafka 一个实例，每个服务器上有一个或多个 kafka 的实例，简单的理解就是一台 kafka 服务器，kafka cluster表示集群的意思
• Topic：消息的主题（类别），可以理解为消息队列，kafka的数据就保存在topic。在每个 broker 上都可以创建多个 topic 。
• Partition：Topic的分区，每个 topic 可以有多个分区，分区的作用是做负载，提高 kafka 的吞吐量。同一个 topic 在不同的分区的数据是不重复的，partition 的表现形式就是一个一个的文件夹！
• Replication：每一个分区都有多个副本，副本的作用是做备胎，主分区（Leader）会将数据同步到从分区（Follower）。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为 Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本
• Message：每一条发送的消息主体。
• Consumer：消费者，即消息的消费方，是消息的出口。
• Consumer Group：我们可以将多个消费者组成一个消费者组，在 kafka 的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量！
• Zookeeper：kafka 集群依赖 zookeeper 来保存集群的的元信息，来保证系统的可用性。

2.1 工作流程

1、发送数据
producer就是生产者，是数据的入口。Producer在写入数据的时候永远的找leader，不会直接将数据写入follower！那leader怎么找呢？写入的流程又是什么样的呢？
我们看下图：

需要注意的一点是，消息写入leader后，follower是主动的去leader进行同步的！producer采用push模式将数据发布到broker，每条消息追加到分区中，顺序写入磁盘，所以保证同一分区内的数据是有序的！写入示意图如下：

分区的主要目的是：
　　1、 方便扩展。因为一个topic可以有多个partition，所以我们可以通过扩展机器去轻松的应对日益增长的数据量。
　　2、 提高并发。以partition为读写单位，可以多个消费者同时消费数据，提高了消息的处理效率。
发送模式

• Kafka提供了发送消息三种模式：（producer.type）
  （oneway、sync、async）
  • 第一种是发送出去就算成功，这种情况当然不能保证消息成功投递到broker；
  • 第二种是Master-Slave（L-F）模型，只有当Master和所有Slave都接收到消息时，才算投递成功，这种模型提供了最高的投递可靠性，但是损伤了性能；
    
  • 第三种模型，即只要Master确认收到消息就算投递成功；实际使用时，根据应用特性选择，绝大多数情况下都会中和可靠性和性能选择第三种模型

生产者配置

KafkaProducer中必填：
bootstrap.servers： 指定生产者客户端连接kafka集群所需的broker地址列表，格式为host1:port1,host2:port2，可以设置一个或多个。这里并非需要所有的broker地址，因为生产者会从给定的broker里寻找其它的broker。

非必填参数：

1. client.id：这个参数用来设定kafkaProducer对应的客户端id，默认值为“”,如果不设置，会自动生成一个非空字符串，内容形式如：“producer-1”,“producer-2”…
2. retries：配置生产者重试次数，对于可重试异常，那么只要在规定的次数内自行恢复了，就不会抛出异常，默认是0。
3. retry.backoff.ms用来设定两次重试之间的时间间隔，默认值100。
4. partitioner.class：显示配置使用哪个分区器。
5. interceptor.classes：指定自定义拦截器，多个传List集合。
6. buffer.memory：生产者客户端RecordAccumulator缓存大小，默认值为33554432B，即32M。
7. batch.size：ProducerBatch可以复用内存区域的大小
   …

2、消费数据
消息存储在log文件后，消费者就可以进行消费了。与生产消息相同的是，消费者在拉取消息的时候也是找leader去拉取。
　　多个消费者可以组成一个消费者组（consumer group），每个消费者组都有一个组id！同一个消费组者的消费者可以消费同一topic下不同分区的数据，但是不会组内多个消费者消费同一分区的数据！！！是不是有点绕。我们看下图：

图示是消费者组内的消费者小于partition数量的情况，所以会出现某个消费者消费多个partition数据的情况，消费的速度也就不及只处理一个partition的消费者的处理速度！如果是消费者组的消费者多于partition的数量，那会不会出现多个消费者消费同一个partition的数据呢？上面已经提到过不会出现这种情况！多出来的消费者不消费任何partition的数据。所以在实际的应用中，建议消费者组的consumer的数量与partition的数量一致！
如果producer的流量增大，当前的topic的parition数量=consumer数量，这时候的应对方式就是横向扩展：增加topic下的partition，同时增加这个consumer group下的consumer。
ConsumerRebalance （重平衡）
本质上是一种协议，规定了一个 Consumer Group 下的所有 consumer 如何达成一致，来分配订阅 Topic 的每个分区。
例如：某 Group 下有 20 个 consumer 实例，它订阅了一个具有 100 个 partition 的 Topic 。正常情况下，kafka 会为每个 Consumer 平均的分配 5 个分区。这个分配的过程就是 Rebalance。

Rebalance 的触发条件有3个:

• 组成员个数发生变化。例如有新的 consumer 加入或者离开组。
• 订阅的 Topic 个数发生变化。
• 订阅 Topic 的分区数发生变化。
  消费者配置
  KafkaConsumer中必填的：
  zookeeper.connect：zookeeper连接服务器地址 （集群可写多个）
  group.id 在使用kafka consumer IPA 的时候，group.id对应点值就是组，这个组是我们自己随便起的，当这个程序启动，这个名字就是组了，其他consumer的group.id也写这个名字，说明这些consumer是同一个组的。

非必选参数

1. zookeeper.session.timeout.ms=5000 zookeeper的session的过期时间
2. zookeeper.connectiontimeout.ms=10000 连接到zk超时时间
3. zookeeper.sync.time.ms=2000 指定多久消费者更新offset到zookeeper中
4. auto.commit.enable=true 自动向zookeeper提交offset信息
5. auto.commit.interval.ms=1000 自动更新时间
6. consumer.id=xxx 当前consumer的标识
   …

2.2 Kafka可靠性保证

• Kakfa Broker Leader的选举：Kakfa Broker集群受Zookeeper管理。所有的Kafka Broker节点一起去Zookeeper上注册一个临时节点，因为只有一个Kafka Broker会注册成功，其他的都会失败，所以这个成功在Zookeeper上注册临时节点的这个Kafka Broker会成为Kafka Broker Controller，其他的Kafka broker叫Kafka Broker follower。（这个过程叫Controller在ZooKeeper注册Watch）。例如：一旦有一个broker宕机了，这个kafka broker controller会读取该宕机broker上所有的partition在zookeeper上的状态，并选取ISR列表中的一个replica作为partition leader（如果ISR列表中的replica全挂，选一个幸存的replica作为leader; 如果该partition的所有的replica都宕机了，则将新的leader设置为-1，等待恢复，等待ISR中的任一个Replica“活”过来，并且选它作为Leader；或选择第一个“活”过来的Replica（不一定是ISR中的）作为Leader），这个broker宕机的事情，kafka controller也会通知zookeeper，zookeeper就会通知其他的kafka broker。
• ISR kafka的ISR机制被称为“不丢消息”机制，涉及到的是kafka的同步，kafka不是完全同步，也不是完全异步。
  首先我们知道kafka 的数据是多副本的。
  每个follower 的数据都是同步leader的，是follower 主动拉取leader 的数据
  Replica的个数小于等于Broker的个数，也就是说，对于每个Partition而言，每个Broker上最多只会有一个Replica，因此可以使用Broker id 指定Partition的Replica

注意：follewer 只是数据的副本提供数据的可恢复性，本身和kafka 的读写性能无关（kafka的读写都是和leader 相关）
isr 的全称是：In-Sync Replicas isr 是一个副本的列表，里面存储的都是能跟leader 数据一致的副本，确定一个副本在isr列表中，有2个判断条件

• 根据副本和leader 的交互时间差，如果大于某个时间差 就认定这个副本不行了，就把此副本从isr 中剔除，此时间差根据
  配置参数rerplica.lag.time.max.ms=10000 也就是默认10s，isr中的follow没有向isr发送心跳包就会被移除
• 根据leader 和副本的信息条数差值决定是否从isr 中剔除此副本，此信息条数差值根据配置参数
  rerplica.lag.max.messages=4000 决定 ，也就是默认消息差大于4000会被移除

注意：在Kafka0.9版本之后,ISR移除了数据条数这一条件,只保留了超时时间这一条件，如果producor一次性发来了10000条数据，而默认条数差立马会大于4000 立马触发ISR

• partition ack说明
  • ack=1，表示producer写partition leader成功后，broker就返回成功，无论其他的partition follower是否写成功。
  • ack=-1[parition的数量]的时候，表示只有producer全部写成功的时候，才算成功，kafka broker才返回成功信息。这里需要注意的是，如果ack=1的时候，一旦有个broker宕机导致partition的follower和leader切换，会导致丢数据。
  • ack=2，表示producer写partition leader和其他一个follower成功的时候，broker就返回成功，无论其他的partition follower是否写成功。
  • At most once：消息可能会丢，但绝不会重复传输。
    At least once：消息绝不会丢，但可能会重复传输。
    Exactly once：每条消息肯定会被传输一次且仅传输一次。
    kafka最多保证At least once，可以保证不丢，但是可能会重复，为了解决重复需要引入唯一标识和去重机制，kafka提供了GUID实现了唯一标识，但是并没有提供自带的去重机制。

2.3 Kafka存储

kafka通过topic来分主题存放数据，主题内有分区，分区可以有多个副本，分区的内部还细分为若干个segment。
所谓的分区其实就是在kafka对应存储目录下创建的文件夹，文件夹的名字是主题名加上分区编号，编号从0开始。

1、segment
所谓的segment其实就是在分区对应的文件夹下产生的文件。
一个分区会被划分成大小相等的若干segment，这样一方面保证了分区的数据被划分到多个文件中保证不会产生体积过大的文件；另一方面可以基于这些segment文件进行历史数据的删除，提高效率。
一个segment又由一个.log和一个.index文件组成。

• .log
  .log文件为数据文件用来存放数据分段数据。
• .index
  .index为索引文件保存对对应的.log文件的索引信息。
  在.index文件中，保存了对对应.log文件的索引信息，通过查找.index文件可以获知每个存储在当前segment中的offset在.log文件中的开始位置，而每条日志有其固定格式，保存了包括offset编号、日志长度、key的长度等相关信息，通过这个固定格式中的数据可以确定出当前offset的结束位置，从而对数据进行读取。
• 命名规则
  这两个文件的命名规则为：
  partition全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值，数值大小为64位，20位数字字符长度，没有数字用0填充。

2、读取数据
开始读取指定分区中某个offset对应的数据时，先根据offset和当前分区的所有segment的名称做比较，确定出数据在哪个segment中，再查找该segment的索引文件，确定当前offset在数据文件中的开始位置，最后从该位置开始读取数据文件，在根据数据格式判断结果，获取完整数据。