[大数据]Kafka架构特性——知识总结

Kafka定义：

• 最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于Hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等

Kafka特性：

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

Kafka使用场景：

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

Kafka设计思想：

• Kafka Broker Leader选举机制：Kafka Broker集群受Zookeeper管理，一个Broker就是一个Kafka服务器，Kafka Broker集群又称为Kafka Cluster，选举机制中，所有的Kafka Broker都会在Zookeeper注册一个临时节点，但只有一个Kafka Broker会注册成功，因此Kafka Broker集群中有一个Kafka Broker Controller其他的为Kafka Broker Follower，HA（一旦有Kafka Broker Controller宕机，Zookeeper上此Broker的临时节点就会消失，其他Broker就会争取在Zookeeper上创建临时节点，产生出新的Kafka Broker Controller），一旦有一个非Kafka Broker Controller的Broker宕机，Kafka Broker Controller会读取该宕机Broker上所有的partition在Zookeeper上的状态，并选取ISR列表中的一个replication 作为partition leader

• Consumer Group & Consumer：

  • 各个Consumer（Consumer线程）可以组成一个Consumer Group，Partition中的每条message只能被Comsumer Group中的一个Consumer消费，如果一个Partition被多个Consumer消费，那么这些Consumer属于不同的Consumer Group，Kafka不支持一个partition中的message由两个或两个以上的同一个consumer group下的consumer thread来处理，除非再启动一个新的consumer group。
  • 当启动一个Consumer Group消费一个Topic时，无论Topic中有多少个Partition，无论Consumer Group中有多少个Consumer，这个Consumer Group下的所有Consumer一定会消费全部的Partition，即便这个Consumer Group中的Consumer只有一个，它也会消费这个Topic中的所有Partition， 如果Consumer Group中的Consumer数量大于Topic中partition的数量，那么Consumer Group中会有Consumer空闲。因此，最优的设计就是，Consumer Group下的Consumer的数量等于Partition数量，这样效率是最高的

• 总结：

  • 一般情况下，一定是一个Consumer Group处理一个Topic的message。Best Practice是这个Consumer Group里面Consumer的数量等于Topic里面Partition的数量，这样效率是最高的，一个Consumer Thread处理一个Partition。如果这个Consumer Group里面Consumer的数量小于Topic里面Partition的数量，就会有Consumer Thread同时处理多个Partition（这个是Kafka自动的机制，我们不用指定），但是总之这个Topic里面的所有Partition都会被处理到的。如果这个Consumer Group里面Consumer的数量大于Topic里面Partition的数量，多出的Consumer Thread就会闲着啥也不干，剩下的是一个Consumer Thread处理一个Partition，这就造成了资源的浪费，因为一个partition不可能被两个Consumer Thread去处理
  • 如果Producer的流量增大，当前的Topic的Parition数量=Consumer数量，这时候的应对方式就是很想扩展：增加Topic下的Partition，同时增加这个Consumer group下的Consumer数量

Topic & Partition:

• Topic相当于传统消息系统MQ中的一个队列Queue，Producer端发送的message必须指定是发送到哪个Topic，但是不需要指定Topic下的哪个Partition，因为Kafka会把收到的message进行load balance，均匀的分布在这个topic下的不同的partition上（ hash(message) % [broker数量]?）。物理上存储上，这个Topic会分成一个或多个Partition，每个Partiton相当于是一个子Queue。在物理结构上，每个Partition对应一个物理的目录（文件夹），文件夹命名是[topicname][partition][序号]，一个Topic可以有无数多的Partition，根据业务需求和数据量来设置

• 一个Topic的Partition数量大于等于Broker的数量，?可以提高吞吐率

• 同一个Partition的Replica尽量分散到不同的机器，?高可用

• 当add a new partition的时候，Partition里面的message不会重新进行分配，原来的Partition里面的message数据不会变，新加的这个Partition刚开始是空的，随后进入这个Topic的message就会重新参与所有Partition的load balance，消息被顺序追加到每个分区日志文件的尾部。Kafka 中的分区可以分布在不同的服务器（broker）上，也就是说，一个主题可以横跨多个broker，以此来提供比单个 broker 更强大的性能

• Partition Replication：一个Topic中的每个Partition可以在其他Broker上存储副本，以便某Broker节点宕机不会影响整个Kafka集群，Replication数量不能大于Cluster中Broker节点的数量，否则报错（这里的Replication数量是Partition的总数，包括一个leader Partition和其他Copy副本的follower Partition）

• 副本数越高，系统虽然越稳定，但是回来带资源和性能上的下降，副本少的话，也会造成系统丢数据的风险

• 传送消息机制：Producer先将message传送给leader Partition，leader Partition再将message复制给follower Partition

• 在向Producer发送ACK前需要保证有多少个Replication已经收到该消息：根据ack配的个数而定

  • Partition Replication中的leader和follower是由Zookeeper进行控制
  • Partition ACK：当ack=0时，表示producer发送数据之后，则就表示成功
  • 当ack=1，表示producer写partition leader成功后，broker就返回成功，无论其他的partition follower是否写成功
  • 当ack=2，表示producer写partition leader和其他一个follower成功的时候， broker就返回成功，无论其他的partition follower是否写成功
  • 当ack=-1 [parition的数量]的时候，表示只有producer全部写成功的时候，才算成功，kafka broker才返回成功信息。这里需要注意的是，如果ack=1的时候，一旦有个broker宕机导致partition的follower和leader切换，会导致丢数据

• Message：

  • Message状态：在Kafka中，消息的状态被保存在consumer中，broker不会关心哪个消息被消费了被谁消费了，只记录一个offset值（指向partition中下一个要被消费的消息位置），这就意味着如果consumer处理不好的话，broker上的一个消息可能会被消费多次
  • Kafka中会把消息持久化到本地文件系统中，并且保持o(1)极高的效率。message的写入持久化是顺序写入的，因此message在被消费的时候也是按顺序被消费的，保证partition的message是顺序消费的
  • Producer：Producer向Topic发送message，可以不需要指定partition，直接发送就好了。kafka通过partition ack来控制是否发送成功并把信息返回给producer
  • （1）At most once消息可能会丢，绝对不会重复传输
  • （2）At least once 消息绝对不会丢，但是可能会重复传输
  • （3）Exactly once每条信息肯定会被传输一次且仅传输一次，这是用户想要的

• push-and-pull?:

  • Kafka中的Producer和consumer采用的是push-and-pull模式，即Producer只管向broker push消息，consumer只管从broker pull消息，两者对消息的生产和消费是异步的

Kafka特点及优势：

• 离线数据装载：Kafka由于对可拓展的数据持久化的支持，它也非常适合向Hadoop或者数据仓库中进行数据装载。
• 实时数据与离线数据：kafka既支持离线数据也支持实时数据，因为kafka的message持久化到文件，并可以设置有效期，因此可以把kafka作为一个高效的存储来使用，可以作为离线数据供后面的分析。当然作为分布式实时消息系统，大多数情况下还是用于实时的数据处理的，但是当cosumer消费能力下降的时候可以通过message的持久化在淤积数据在kafka。
• 解耦:?相当于一个MQ，使得Producer和Consumer之间异步的操作，系统之间解耦
• 冗余:?replication有多个副本，保证一个broker node宕机后不会影响整个服务
• 扩展性:?broker节点可以水平扩展，partition也可以水平增加，partition replica也可以水平增加
• 峰值:?在访问量剧增的情况下，kafka水平扩展, 应用仍然需要继续发挥作用
• 可恢复性:?系统的一部分组件失效时，由于有partition的replica副本，不会影响到整个系统。
• 顺序保证性：由于kafka的producer的写message与consumer去读message都是顺序的读写，保证了高效的性能。
• 缓冲：由于producer那面可能业务很简单，而后端consumer业务会很复杂并有数据库的操作，因此肯定是producer会比consumer处理速度快，如果没有kafka，producer直接调用consumer，那么就会造成整个系统的处理速度慢，加一层kafka作为MQ，可以起到缓冲的作用。
• 异步通信：作为MQ，Producer与Consumer异步通信

Kafka集群例子：

• 一个典型的Kafka集群中包含若干Producer（可以是web前端FET，或者是服务器日志等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干ConsumerGroup，以及一个Zookeeper集群。Kafka通过Zookeeper管理Kafka集群配置：选举Kafka broker的leader，以及在Consumer Group发生变化时进行rebalance，因为consumer消费kafka topic的partition的offsite信息是存在Zookeeper的。Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费消息

Kafka零拷贝：

• 定义：所谓的零拷贝是指将数据直接从磁盘文件复制到网卡设备中，而不需要经由应用程序之手。零拷贝大大提高了应用程序的性能，减少了内核和用户模式之间的上下文切换
• 原4次拷贝过程：你需要将静态内容（类似图片、文件）展示给用户。这个情形就意味着需要先将静态内容从磁盘中复制出来放到一个内存 buf 中，然后将这个 buf 通过套接字（Socket）传输给用户，进而用户获得静态内容。这看起来再正常不过了，但实际上这是很低效的流程
• 技术效果：零拷贝技术，那么应用程序可以直接请求内核把磁盘中的数据传输给 Socket

Kafka性能为什么好：

• Kafka不基于内存，而是磁盘存储，因此消息堆积能力更强
• 顺序写入：利用磁盘的顺序访问速度可以接近内存，Kafka的消息都是append操作，写入partition是有序的，节省了磁盘的「寻道时间」
• 传统拷贝：
  1. 读取磁盘文件数据到「内核缓冲区」
  2. 将「内核缓冲区」的数据拷贝到「用户缓冲区」
  3. 将「用户缓冲区」的数据拷贝到「socket发送缓冲区」
  4. 将「socket发送缓冲区」的数据发送到网卡，进行数据传输

Zookeeper在Kafka中的角色：

• Broker在Zookeeper中注册：临时节点，本质上就是在 ZooKeeper 中创建一个专属的目录（又称为节点），其路径为?/ brokers，存放broker.name、端口号

• Topic在Zookeeper中注册：协助leader partition的选举，/ brokers / topics / {topic_name}

• Consumer Group在Zookeeper中注册：?/ consumers / {group_id}，有3个子目录

  1. ids：Consumer Group 中有多个 Consumer，ids 用于记录这些 Consumer

  2. owners：记录该Consumer Group可消费的Topic信息

  3. offsets：记录owners中每个Topic的所有Partition的Offset

• Consumer在Zookeeper中注册：/ consumers / {group_id} / ids下

• Consumer负载均衡：一个 Consumer Group 中有多个 Consumer，让它们尽可能均匀地消费订阅的消息

• Producer负载均衡：一个 Topic 的 Partition 会尽量分散到不同的 Broker 上。而 Producers 则根据指定的 Topic 将消息 Push 到相应的 Partition，让消息均衡地 Push 到各个 Partition