| 编辑推荐: |
本文首先介绍Kafka的原理，
Kafka集群分类存储以及Kafka的核心抽象概念记录流，然后详细阐述Kafka的应用场景如何作为流应用平台以及数据流处理，希望对您的学习有所帮助。
本文来自于网络大数据-技术分享，由火龙果软件Alice编辑、推荐。
|
|
1、简介
它可以让你发布和订阅记录流。在这方面，它类似于一个消息队列或企业消息系统。
它可以让你持久化收到的记录流，从而具有容错能力。
首先，明确几个概念：
Kafka运行在一个或多个服务器上。
Kafka集群分类存储的记录流被称为主题(Topics)。
每个消息记录包含一个键，一个值和时间戳。
Kafka有四个核心API：
生产者 API 允许应用程序发布记录流至一个或多个Kafka的话题(Topics)。
消费者API 允许应用程序订阅一个或多个主题，并处理这些主题接收到的记录流。
Streams API 允许应用程序充当流处理器(stream processor)，从一个或多个主题获取输入流，并生产一个输出流至一个或多个的主题，能够有效地变换输入流为输出流。
Connector API 允许构建和运行可重用的生产者或消费者，能够把 Kafka主题连接到现有的应用程序或数据系统。例如，一个连接到关系数据库的连接器(connector)可能会获取每个表的变化。
Kafka的客户端和服务器之间的通信是靠一个简单的，高性能的，与语言无关的TCP协议完成的。这个协议有不同的版本，并保持向前兼容旧版本。Kafka不光提供了一个Java客户端，还有许多语言版本的客户端。
2、 架构
2.1 Broker
每个kafka server称为一个Broker，多个borker组成kafka cluster。一个机器上可以部署一个或者多个Broker，这多个Broker连接到相同的ZooKeeper就组成了Kafka集群。
2.2 主题Topic
让我们先来了解Kafka的核心抽象概念记录流 – 主题。主题是一种分类或发布的一系列记录的名义上的名字。Kafka的主题始终是支持多用户订阅的;
也就是说，一个主题可以有零个，一个或多个消费者订阅写入的数据。
Topic 与broker
一个Broker上可以创建一个或者多个Topic。同一个topic可以在同一集群下的多个Broker中分布。
当然，Topic只是一个名义上的组件，真正在Broker间分布式的Partition。
2.3 分区与日志
一个主题对应多个分区，一个分区对应一个日志
Kafka会为每个topic维护了多个分区(partition)，每个分区会映射到一个逻辑的日志(log)文件。每个分区是一个有序的，不可变的消息序列，新的消息不断追加到这个有组织的有保证的日志上。分区会给每个消息记录分配一个顺序ID号
– 偏移量， 能够唯一地标识该分区中的每个记录。
日志分区是分布式的存在于一个kafka集群的多个broker上。每个partition会被复制多份存在于不同的broker上。这样做是为了容灾。具体会复制几份，会复制到哪些broker上，都是可以配置的。经过相关的复制策略后，每个topic在每个broker上会驻留一到多个partition：
2.4 保留策略与Offset
Kafka集群保留所有发布的记录，不管这个记录有没有被消费过，Kafka提供可配置的保留策略去删除旧数据(还有一种策略根据分区大小删除数据)。例如，如果将保留策略设置为两天，在记录公布后两天内，它可用于消费，之后它将被丢弃以腾出空间。Kafka的性能跟存储的数据量的大小无关，
所以将数据存储很长一段时间是没有问题的。
事实上，保留在每个消费者元数据中的最基础的数据就是消费者正在处理的当前记录的偏移量(offset)或位置(position)。这种偏移是由消费者控制：通常偏移会随着消费者读取记录线性前进，但事实上，因为其位置是由消费者进行控制，消费者可以在任何它喜欢的位置读取记录。例如，消费者可以恢复到旧的偏移量对过去的数据再加工或者直接跳到最新的记录，并消费从“现在”开始的新的记录。
这些功能的结合意味着，实现Kafka的消费者的代价都是很小的，他们可以增加或者减少而不会对集群或其他消费者有太大影响。例如，你可以使用我们的命令行工具去追随任何主题，而且不会改变任何现有的消费者消费的记录。
2.5 Leader与Followers
一个Topic可能有很多分区，以便它能够支持海量的的数据，更重要的意义是分区是进行并行处理的基础单元。日志的分区会跨服务器的分布在Kafka集群中，每个分区可以配置一定数量的副本分区提供容错能力。为了保证较高的处理效率，消息的读写都是在固定的一个副本上完成。这个副本就是所谓的Leader，而其他副本则是Follower，而Follower则会定期地到Leader上同步数据。
(1)leader处理所有的读取和写入分区的请求，而followers被动的从领导者拷贝数据。
(2)如果leader失败了，followers之一将自动成为新的领导者。
(3)每个服务器可能充当一些分区的leader和其他分区的follower，这样的负载就会在集群内很好的均衡分配。
(4)一个分区在同一时刻只能有一个消费者实例进行消费。
举例：
可以看见我们一共有3个分区分别是0，1，2， replica 有2个：
partition 0 的leader在broker1， follower在broker2
partition 1 的leader在broker2， follower在broker0
partition 2 的leader在broker0， follower在brokder1
一个broker中不会出现两个一样的Partition，replica会被均匀的分布在各个kafka
server(broker)上 。Kafka并不允许replicas 数设置大于 broker数，因为在一个broker上如果有2个replica其实是没有意义的，因为再多的replica同时在一台broker上，随着该broker的crash，一起不可用。
(1)Leader选举与ISR
如果某个分区所在的服务器除了问题，不可用，kafka会从该分区的其他的副本中选择一个作为新的Leader。之后所有的读写就会转移到这个新的Leader上。现在的问题是应当选择哪个作为新的Leader。显然，只有那些跟Leader保持同步的Follower才应该被选作新的Leader。
Kafka会在Zookeeper上针对每个Topic维护一个称为ISR(in-sync replica，已同步的副本)的集合，该集合中是一些分区的副本。只有当这些副本都跟Leader中的副本同步了之后，kafka才会认为消息已提交，并反馈给消息的生产者。如果这个集合有增减，kafka会更新zookeeper上的记录。如果某个分区的Leader不可用，Kafka就会从ISR集合中选择一个副本作为新的Leader。显然通过ISR，kafka需要的冗余度较低，可以容忍的失败数比较高。假设某个topic有f+1个副本，kafka可以容忍f个服务器不可用。
(2)为什么不用少数服从多数的方法
少数服从多数是一种比较常见的一致性算法和Leader选举法。它的含义是只有超过半数的副本同步了，系统才会认为数据已同步;选择Leader时也是从超过半数的同步的副本中选择。这种算法需要较高的冗余度。譬如只允许一台机器失败，需要有三个副本;而如果只容忍两台机器失败，则需要五个副本。而kafka的ISR集合方法，分别只需要两个和三个副本。
(3)如果所有的ISR副本都失败了怎么办
此时有两种方法可选，一种是等待ISR集合中的副本复活，一种是选择任何一个立即可用的副本，而这个副本不一定是在ISR集合中。这两种方法各有利弊，实际生产中按需选择。如果要等待ISR副本复活，虽然可以保证一致性，但可能需要很长时间。而如果选择立即可用的副本，则很可能该副本并不一致。
2.6 生产者和消费者
(1)生产者
生产者发布数据到他们所选择的主题。生产者负责选择把记录分配到主题中的哪个分区。这可以使用轮询算法(
round-robin)进行简单地平衡负载，也可以根据一些更复杂的语义分区算法(比如基于记录一些键值)来完成。
(2)消费者
消费者以消费群(consumer group)的名称来标识自己，每个发布到主题的消息都会发送给订阅了这个主题的消费群里面的一个消费者实例，即一个消费群只发送一次。消费者的实例可以在单独的进程或单独的机器上。
上图中两个服务器的Kafka集群具有四个分区(P0-P3)和两个消费群。A消费群有两个消费者，B群有四个。更常见的是，我们会发现主题有少量的消费群，每一个都是“逻辑上的订阅者”。每组都是由很多消费者实例组成，从而实现可扩展性和容错性。这只不过是发布
– 订阅模式的再现，区别是这里的订阅者是一组消费者而不是一个单一的进程的消费者。
Kafka消费群的实现方式是通过分割分区给每个Consumer实例实现的，使每个实例在任何时间点的都可以“公平分享”独占的分区。维持消费群中的成员关系的这个过程是通过Kafka动态协议处理。如果新的实例加入该组，他将接管该组的其他成员的一些分区;
如果一个实例死亡，其分区将被分配到剩余的实例。
Kafka只保证一个分区内的消息有序，不能保证一个主题的不同分区之间的消息有序。分区的消息有序与依靠主键进行数据分区的能力相结合足以满足大多数应用的要求。但是，如果你想要保证所有的消息都绝对有序可以只为一个主题分配一个分区，虽然这将意味着每个消费群同时只能有一个消费进程在消费。
3 、数据可靠性与一致性
3.1 Partition Recovery机制
每个Partition会在磁盘记录一个RecoveryPoint，记录已经flush到磁盘的最大offset。当broker
fail 重启时，会进行loadLogs。 首先会读取该Partition的RecoveryPoint，找到包含RecoveryPoint的segment及以后的segment，
这些segment就是可能没有完全flush到磁盘segments。然后调用segment的recover，重新读取各个segment的msg，并重建索引。
优点
以segment为单位管理Partition数据，方便数据生命周期的管理，删除过期数据简单。
在程序崩溃重启时，加快recovery速度，只需恢复未完全flush到磁盘的segment。
通过index中offset与物理偏移映射，用二分查找能快速定位msg，并且通过分多个Segment，每个index文件很小，查找速度更快。
3.2 Partition Replica同步机制
Partition的多个replica中一个为Leader，其余为follower
Producer只与Leader交互，把数据写入到Leader中
Followers从Leader中拉取数据进行数据同步
Consumer只从Leader拉取数据
ISR：in-sync replica，已同步的副本。准确的定义是“所有不落后的replica集合”。不落后有两层含义:距离上次FetchRequest的时间不大于某一个值或落后的消息数不大于某一个值，
Leader失败后会从ISR中选取一个Follower做Leader。
3.3 消息的顺序消费问题
在说到消息中间件的时候，我们通常都会谈到一个特性：消息的顺序消费问题。这个问题看起来很简单：Producer发送消息1,
2, 3;Consumer按1, 2, 3顺序消费。但实际情况却是：无论RocketMQ，还是Kafka，缺省都不保证消息的严格有序消费!困难如下：
(1)Producer
发送端不能异步发送，异步发送在发送失败的情况下，就没办法保证消息顺序。比如你连续发了1，2，3。
过了一会，返回结果1失败，2, 3成功。你把1再重新发送1遍，这个时候顺序就乱掉了。
(2)存储端
对于存储端，要保证消息顺序，会有以下几个问题：
消息不能分区。也就是1个topic，只能有1个队列。在Kafka中，它叫做partition;在RocketMQ中，它叫做queue。
如果你有多个队列，那同1个topic的消息，会分散到多个分区里面，自然不能保证顺序。
即使只有1个队列的情况下，会有第2个问题。该机器挂了之后，能否切换到其他机器也就是高可用问题。比如你当前的机器挂了，上面还有消息没有消费完。此时切换到其他机器，可用性保证了。但消息顺序就乱掉了。要想保证，一方面要同步复制，不能异步复制;另1方面得保证，切机器之前，挂掉的机器上面，所有消息必须消费完了，不能有残留。很明显，这个很难。
(3)接收端
对于接收端，不能并行消费，也即不能开多线程或者多个客户端消费同1个队列。
3.4 Producer发送消息的配置
3.4.1 同步模式
kafka有同步(sync)、异步(async)以及oneway这三种发送方式，某些概念上区分也可以分为同步和异步两种，同步和异步的发送方式通过producer.type参数指定，而oneway由request.require.acks参数指定。
producer.type的默认值是sync，即同步的方式。这个参数指定了在后台线程中消息的发送方式是同步的还是异步的。如果设置成异步的模式，可以运行生产者以batch的形式push数据，这样会极大的提高broker的性能，但是这样会增加丢失数据的风险。
3.4.2 异步模式
对于异步模式，还有4个配套的参数，如下：
3.4.3 oneway
oneway是只顾消息发出去而不管死活，消息可靠性最低，但是低延迟、高吞吐，这种对于某些完全对可靠性没有要求的场景还是适用的，即request.required.acks设置为0。
3.4.4 消息可靠性级别
当Producer向Leader发送数据时，可以通过request.required.acks参数设置数据可靠性的级别：
0: 不论写入是否成功，server不需要给Producer发送Response，如果发生异常，server会终止连接，触发Producer更新meta数据;
1: Leader写入成功后即发送Response，此种情况如果Leader fail，会丢失数据
-1: 等待所有ISR接收到消息后再给Producer发送Response，这是最强保证
仅设置acks=-1也不能保证数据不丢失，当Isr列表中只有Leader时，同样有可能造成数据丢失。要保证数据不丢除了设置acks=-1，
还要保 证ISR的大小大于等于2，具体参数设置:
(1)request.required.acks: 设置为-1 等待所有ISR列表中的Replica接收到消息后采算写成功;
(2)min.insync.replicas: 设置为大于等于2，保证ISR中至少有两个Replica
Producer要在吞吐率和数据可靠性之间做一个权衡。
3.4.5 一般配置
4、 应用场景
4.1 消息系统
消息处理模型历来有两种：
队列模型：一组消费者可以从服务器读取记录，每个记录都会被其中一个消费者处理，为保障消息的顺序，同一时刻只能有一个进程进行消费。
发布-订阅模型：记录被广播到所有的消费者。
Kafka的消费群的推广了这两个概念。消费群可以像队列一样让消息被一组进程处理(消费群的成员)，与发布
– 订阅模式一样，Kafka可以让你发送广播消息到多个消费群。
Kafka兼顾了消息的有序性和并发处理能力。传统的消息队列的消息在队列中是有序的，多个消费者从队列中消费消息，服务器按照存储的顺序派发消息。然而，尽管服务器是按照顺序派发消息，但是这些消息记录被异步传递给消费者，消费者接收到的消息也许已经是乱序的了。这实际上意味着消息的排序在并行消费中都将丢失。消息系统通常靠
“排他性消费”( exclusive consumer)来解决这个问题，只允许一个进程从队列中消费，当然，这意味着没有并行处理的能力。
Kafka做的更好。通过一个概念：并行性-分区-主题实现主题内的并行处理，Kafka是能够通过一组消费者的进程同时提供排序保证和并行处理以及负载均衡的能力：
(1)排序保障
每个主题的分区指定给每个消费群中的一个消费者，使每个分区只由该组中的一个消费者所消费。通过这样做，我们确保消费者是一个分区唯一的读者，从而顺序的消费数据。
(2)并行处理
因为有许多的分区，所以负载还能够均衡的分配到很多的消费者实例上去。但是请注意，一个消费群的消费者实例不能比分区数量多，因为分区数代表了一个主题的最大并发数，消费者的数量高于这个数量意义不大。
4.2 日志采集
大多数时候，我们的log都会输出到本地的磁盘上，排查问题也是使用linux命令来搞定，如果web程序组成负载集群，那么就有多台机器，如果有几十台机器，几十个服务，那么想快速定位log问题和排查就比较麻烦了，所以很有必要有一个统一的平台管理log，现在大多数公司的套路都是收集重要应用的log集中到kafka中，然后在分别导入到es和hdfs上，一个做实时检索分析，另一个做离线统计和数据备份。如何能快速收集应用日志到kafka中
方法一：使用log4j的集成包
kafka官网已经提供了非常方便的log4j的集成包 kafka-log4j-appender，我们只需要简单配置log4j文件，就能收集应用程序log到kafka中。
注意，需要引入maven的依赖包：
非常简单，一个maven依赖加一个log4j配置文件即可，如果依然想写入log到本地 文件依然也是可以的，这种方式最简单快速，但是默认的的log日志是一行一行的纯文本，有些场景下我们可能需要json格式的数据。
方法二: 重写Log4jAppender
重写Log4jAppender，自定义输出格式，支持json格式，如果是json格式的数据打入到kafka中，后续收集程序可能就非常方便了，直接拿到json就能入到mongodb或者es中，如果打入到kafka中的数据是纯文本，那么收集程序，可能需要做一些etl，解析其中的一些字段然后再入到es中，所以原生的输出格式，可能稍不灵活，这样就需要我们自己写一些类，然后达到灵活的程度，github连接：https://github.com/qindongliang/log_to_kafka感兴趣的朋友可以看下。
总结：
(1)方法一简单快速，不支持json格式的输出，打到kafka的消息都是原样的log日志信息
(2)方法二稍微复杂，需要自己扩展log收集类，但支持json格式的数据输出，对于想落地json数据直接到存储系统中是非常适合的。
此外需要注意，在调试的时候log发送数据到kafka模式最好是同步模式的否则你控制台打印的数据很有可能不会被收集kafka中，程序就停止了。生产环境最好开启异步发送数据模式，因为内部是批量的处理，所以能提升吞吐,但有一定的轻微延迟。
4.3 流处理
只是读，写，以及储存数据流是不够的，目的是能够实时处理数据流。在Kafka中，流处理器是从输入的主题连续的获取数据流，然后对输入进行一系列的处理，并生产连续的数据流到输出主题。
这些简单处理可以直接使用生产者和消费者的API做到。然而，对于更复杂的转换Kafka提供了一个完全集成的流API。这允许应用程序把一些重要的计算过程从流中剥离或者加入流一起。这种设施可帮助解决这类应用面临的难题：处理杂乱的数据，改变代码去重新处理输入，执行有状态的计算等。流API建立在Kafka提供的核心基础单元之上：它使用生产者和消费者的API进行输入输出，使用Kafka存储有状态的数据，并使用群组机制在一组流处理实例中实现容错。
把功能组合起来
消息的传输，存储和流处理的组合看似不寻常，却是Kafka作为流处理平台的关键。像HDFS分布式文件系统，允许存储静态文件进行批量处理。像这样的系统允许存储和处理过去的历史数据。传统的企业消息系统允许处理您订阅后才抵达的消息。这样的系统只能处理将来到达的数据。
Kafka结合了这些功能，这种结合对Kafka作为流应用平台以及数据流处理的管道至关重要。通过整合存储和低延迟订阅，流处理应用可以把过去和未来的数据用相同的方式处理。这样一个单独的应用程序，不但可以处理历史的，保存的数据，当它到达最后一条记录不会停止，继续等待处理未来到达的数据。这是泛化了的流处理的概念，包括了批处理应用以及消息驱动的应用。同样，流数据处理的管道结合实时事件的订阅使人们能够用Kafka实现低延迟的管道;
可靠的存储数据的能力使人们有可能使用它传输一些重要的必须保证可达的数据。可以与一个定期加载数据的线下系统集成，或者与一个因为维护长时间下线的系统集成。流处理的组件能够保证转换(处理)到达的数据。
5、Kafka与ActiveMQ对比
首先，Active MQ与Kafka的相同点只有一个，就是都是消息中间件。其他没有任何相同点。
5.1 consumer的不同
(1)AMQ消费完的消息会被清理掉
AMQ无论在standalone还是分布式的情况下，都会使用mysql作为存储，多一个consumer线程去消费多个queue,
消费完的message会在mysql中被清理掉。
(2)AMQ的消费逻辑在Broker中完成
作为AMQ的consume clinet的多个consumer线程去消费queue，AMQ Broker会接收到这些consume线程，阻塞在这里，有message来了就会进行消费，没有消息就会阻塞在这里。具体消费的逻辑也就是处理这些consumer线程都是AMQ
Broker那面处理。
kafka是message都存在partition下的segment文件里面，有offsite偏移量去记录那条消费了，哪条没消费。某个consumer
group下consumer线程消费完就会，这个consumer group 下的这个consumer对应这个partition的offset+1，kafka并不会删除这条已经被消费的message。其他的consumer
group也可以再次消费这个message。在high level api中offset会自动或手动的提交到zookeeper上(如果是自动提交就有可能处理失败或还没处理完就提交offset+1了，容易出现下次再启动consumer
group的时候这条message就被漏了)，也可以使用low level api，那么就是consumer程序中自己维护offset+1的逻辑。kafka中的message会定期删除。
(3)Kafka有consumer group的概念，AMQ没有。
一个consumer group下有多个consumer，每个consumer都是一个线程，consumer
group是一个线程组。每个线程组consumer group之间互相独立。同一个partition中的一个message只能被一个consumer
group下的一个consumer线程消费，因为消费完了这个consumer group下的这个consumer对应的这个partition的offset就+1了，这个consumer
group下的其他consumer还是这个consumer都不能在消费了。 但是另外一个consumer
group是完全独立的，可以设置一个from的offset位置，重新消费这个partition。
5.2 关于存储结构
ActiveMQ的消息持久化机制有JDBC，AMQ，KahaDB和LevelDB
Kafka是文件存储，每个topic有多个partition，每个partition有多个replica副本(每个partition和replica都是均匀分配在不同的kafka
broker上的)。每个partition由多个segment文件组成。这些文件是顺序存储的。因此读取和写入都是顺序的，因此，速度很快，省去了磁盘寻址的时间。
很多系统、组件为了提升效率一般恨不得把所有数据都扔到内存里，然后定期flush到磁盘上;而Kafka决定直接使用页面缓存;但是随机写入的效率很慢，为了维护彼此的关系顺序还需要额外的操作和存储，而线性的顺序写入可以避免磁盘寻址时间，实际上，线性写入(linear
write)的速度大约是300MB/秒，但随即写入却只有50k/秒，其中的差别接近10000倍。这样，Kafka以页面缓存为中间的设计在保证效率的同时还提供了消息的持久化，每个consumer自己维护当前读取数据的offset(也可委托给zookeeper)，以此可同时支持在线和离线的消费。
5.3 关于使用场景与吞吐量
ActiveMQ用于企业消息中间件，使得业务逻辑和前端处理逻辑解耦。AMQ的吞吐量不大，zuora的AMQ就是用作jms来使用。AMQ吞吐量不够，并且持久化message数据通过jdbc存在mysql，写入和读取message性能太低。而Kafka的吞吐量非常大。
5.4 push/pull 模型
对于消费者而言有两种方式从消息中间件获取消息：
①Push方式：由消息中间件主动地将消息推送给消费者，采用Push方式，可以尽可能快地将消息发送给消费者;②Pull方式：由消费者主动向消息中间件拉取消息，会增加消息的延迟，即消息到达消费者的时间有点长
但是，Push方式会有一个坏处：如果消费者的处理消息的能力很弱(一条消息需要很长的时间处理)，而消息中间件不断地向消费者Push消息，消费者的缓冲区可能会溢出。
AMQ的Push消费
ActiveMQ使用PUSH模型， 对于PUSH，broker很难控制数据发送给不同消费者的速度。AMQ
Broker将message推送给对应的BET consumer。ActiveMQ用prefetch
limit 规定了一次可以向消费者Push(推送)多少条消息。当推送消息的数量到达了perfetch
limit规定的数值时，消费者还没有向消息中间件返回ACK，消息中间件将不再继续向消费者推送消息。
AMQ的Pull消费
ActiveMQ prefetch limit 设置成0意味着什么意味着此时，消费者去轮询消息中间件获取消息。不再是Push方式了，而是Pull方式了。即消费者主动去消息中间件拉取消息。
那么，ActiveMQ中如何采用Push方式或者Pull方式呢从是否阻塞来看，消费者有两种方式获取消息。同步方式和异步方式。
同步方式使用的是ActiveMQMessageConsumer的receive()方法。而异步方式则是采用消费者实现MessageListener接口，监听消息。使用同步方式receive()方法获取消息时，prefetch
limit即可以设置为0，也可以设置为大于0。
prefetch limit为零 意味着：“receive()方法将会首先发送一个PULL指令并阻塞，直到broker端返回消息为止，这也意味着消息只能逐个获取(类似于Request<->Response)”。
prefetch limit 大于零 意味着：“broker端将会批量push给client 一定数量的消息(<=
prefetch)，client端会把这些消息(unconsumedMessage)放入到本地的队列中，只要此队列有消息，那么receive方法将会立即返回，当一定量的消息ACK之后，broker端会继续批量push消息给client端。”
当使用MessageListener异步获取消息时，prefetch limit必须大于零了。因为，prefetch
limit 等于零 意味着消息中间件不会主动给消费者Push消息，而此时消费者又用MessageListener被动获取消息(不会主动去轮询消息)。这二者是矛盾的。
Kafka只有Pull消费方式
Kafka使用PULL模型，PULL可以由消费者自己控制，但是PULL模型可能造成消费者在没有消息的情况下盲等，这种情况下可以通过long
polling机制缓解，而对于几乎每时每刻都有消息传递的流式系统，这种影响可以忽略。Kafka 的
consumer 是以pull的形式获取消息数据的。 pruducer push消息到kafka cluster
，consumer从集群中pull消息。
|
