消息队列之kafka 入门

[TOC]

Kafka基本概念

Kafka是一个分布式流数据系统，使用Zookeeper进行集群的管理。与其他消息系统类似，整个系统由生产者、Broker Server和消费者三部分组成，生产者和消费者由开发人员编写，通过API连接到Broker Server进行数据操作。我们重点关注三个概念：

Kafka部分名词解释如下：

• Broker：消息中间件处理结点，一个Kafka节点就是一个broker，多个broker可以组成一个Kafka集群。
• Topic：一类消息，例如page view日志、click日志等都可以以topic的形式存在，Kafka集群能够同时负责多个topic的分发。
• Partition：topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。

Topic

Topic，是Kafka下消息的类别，相当于一条专用的消息通道。对于大多数人来说，在开发的时候只需要关注数据写入到了哪个topic、从哪个topic取出数据。与 pulsar 对比，kafka 没有租户和命名空间的概念，直接就是一个 topic。

Topic 就是数据主题，是数据记录发布的地方,可以用来区分业务系统。Kafka中的Topics总是多订阅者模式，一个topic可以拥有一个或者多个消费者来订阅它的数据。

对于每一个topic， Kafka集群都会维持一个分区日志，如下所示：

每个分区都是有序且顺序不可变的记录集，并且不断地追加到结构化的commit log文件。分区中的每一个记录都会分配一个id号来表示顺序，我们称之为offset，offset用来唯一的标识分区中每一条记录。

Kafka 集群保留所有发布的记录—无论他们是否已被消费—并通过一个可配置的参数——保留期限来控制. 举个例子， 如果保留策略设置为2天，一条记录发布后两天内，可以随时被消费，两天过后这条记录会被抛弃并释放磁盘空间。Kafka的性能和数据大小无关，所以长时间存储数据没有什么问题.

事实上，在每一个消费者中唯一保存的元数据是offset（偏移量） 即消费在log中的位置.偏移量由消费者所控制:通常在读取记录后，消费者会以线性的方式增加偏移量，但是实际上，由于这个位置由消费者控制，所以消费者可以采用任何顺序来消费记录。例如，一个消费者可以重置到一个旧的偏移量，从而重新处理过去的数据；也可以跳过最近的记录，从”现在”开始消费。

这些细节说明Kafka 消费者是非常廉价的—消费者的增加和减少，对集群或者其他消费者没有多大的影响。比如，你可以使用命令行工具，对一些topic内容执行 tail操作，并不会影响已存在的消费者消费数据。

日志中的 partition（分区）有以下几个用途。第一，当日志大小超过了单台服务器的限制，允许日志进行扩展。每个单独的分区都必须受限于主机的文件限制，不过一个主题可能有多个分区，因此可以处理无限量的数据。第二，可以作为并行的单元集—关于这一点，更多细节如下

Partition

Partition，是Kafka下数据存储的基本单元，这个是物理上的概念。同一个topic的数据，会被分散的存储到多个partition中，这些partition可以在同一台机器上，也可以是在多台机器上，比如下图所示的topic就有4个partition，分散在两台机器上。这种方式在大多数分布式存储中都可以见到，比如MongoDB、Elasticsearch的分片技术，其优势在于：有利于水平扩展，避免单台机器在磁盘空间和性能上的限制，同时可以通过复制来增加数据冗余性，提高容灾能力。为了做到均匀分布，通常partition的数量通常是Broker Server数量的整数倍。partition 就是 kafka 支持支持消息和消费者横向扩展的基础

每个分区都有一台 server 作为 “leader”，零台或者多台server作为 follwers 。leader server 处理一切对 partition （分区）的读写请求，而follwers只需被动的同步leader上的数据。当leader宕机了，followers 中的一台服务器会自动成为新的 leader。每台 server 都会成为某些分区的 leader 和某些分区的 follower，因此集群的负载是平衡的。

Consumer Group

Consumer Group，同样是逻辑上的概念，是Kafka实现单播和广播两种消息模型的手段。同一个topic的数据，会广播给不同的group；同一个group中的worker，只有一个worker能拿到这个数据。换句话说，对于同一个topic，每个group都可以拿到同样的所有数据，但是数据进入group后只能被其中的一个worker消费。group内的worker可以使用多线程或多进程来实现，也可以将进程分散在多台机器上，worker的数量通常不超过partition的数量，且二者最好保持整数倍关系，因为Kafka在设计时假定了一个partition只能被一个worker消费（同一group内）。这里其实已经说得很清楚了。对于一个 group 内的多个worker 来看，就是单播， 对于多个 group 来看，就是广播。与 pulsar 作为对比，不同的 group，相当于 Shared 模式下，不同的 subScribetionName。同一个 group 内的多个 worker， 相当于 Shared 模式下，配置了相同的 subScribetionName的 多个 worker。

分组消费再平衡策略

当一个group中,有consumer加入或者离开时,会触发partitions均衡partition.assignment.strategy,决定了partition分配给消费者的分配策略，有两种分配策略：

org.apache.kafka.clients.consumer.RangeAssignor

默认采用的是这种再平衡方式，这种方式分配只是针对消费者订阅的topic的单个topic所有分区再分配，Consumer Rebalance的算法如下：

• A=(partition数量/同分组消费者总个数)
• M=对上面所得到的A值小数点第一位向上取整
• 计算出该消费者拉取数据的patition合集：Ci = [P(M*i ),P((i + 1) * M -1)]

例如 partition 有 6个，消费者有8个，他们属于同一个消费组。

A=6/8=0.75
M=1
C0=[P(1*0),P((0+1)*1-1)]=[P0,P0] 
C1=[P(1*1),P((1+1)*1-1)]=[P1,P1] 
C2=[P(1*2),P((2+1)*1-1)]=[P2,P2] 
C3=[P(1*3),P((3+1)*1-1)]=[P3,P3] 
C4=[P(1*4),P((4+1)*1-1)]=[P4,P4] 
C5=[P(1*5),P((5+1)*1-1)]=[P5,P5] 
C6=[P(1*6),P((6+1)*1-1)]=[P6,P6] 
C7=[P(1*7),P((7+1)*1-1)]=[P7,P7]

# 但是partition只有P0-P5根本就没有P6和P7，所以这两个消费者相当于是会被闲置的，就相当于占用资源，却没什么用，所以在这里真正起到作用的就是C0-C5。

例如 partition 有 6个，消费者有5个，他们属于同一个消费组。

A=6/5=1.2 
M=2

C0=[P(2*0),P((0+1)*2-1)]=[P0,P1] 
C1=[P(2*1),P((1+1)*2-1)]=[P2,P3] 
C2=[P(2*2),P((2+1)*2-1)]=[P4,P5]
C3=[P(2*3),P((3+1)*2-1)]=[P6,P7] 
C4=[P(2*4),P((4+1)*2-1)]=[P8,P9]
# 同上面一样C3和C4没有起到任何作用。

小结

• 按照如上的算法，所以如果kafka的消费组需要增加组员，最多增加到和partition数量一致，超过的组员只会占用资源，而不起作用；
• kafka的partition的个数一定要大于消费组组员的个数，并且partition的个数对于消费组组员取模一定要为0，不然有些消费者会占用资源却不起作用；

分组成员的存活检测

分组消费有一个比较好的功能就是自动检测失败的消费者并将其踢出分组，然后重新进行分区分配。那么kafka是如何检测失败的消费者的呢。我们就拿0.10.x为例进行讲解说明。

消费着订阅了一组的topic后，会在调用poll(long)函数的时候加入分组，分组内新增消费者就会进行再平衡。Poll 函数的设计目标就是来保证消费者存活的。只要持续不断的调用poll函数，消费者就会留在分组里，连续的从分配给他的分区里消费消息。

存活检测 (live)

消费者也会使用一个后台线程发送周期性的心跳给broker。如果消费者挂掉或者无法在session.timeout.ms时间范围内发送心跳，消费者会被视为死亡，它的分区就会被重新分配。心跳检测能保证消费者存活。

活跃检测 (active)

由于心跳是后台线程周期性发送的，那么会存在消费者心跳正常发送，但是不消费消息的情况。为了避免这种消费者无限期的占用分配给他的分区这种情况，kafka提供了一种活跃检测机制，使用max.poll.interval.ms配置。根本上来说，两次调用poll函数的间隔大于该值，消费者就会离开分组，然后它的分区会被其它消费着消费。当发生这种情况时，你会收到一个offset提交失败的异常。这种机制确保了只有活跃的消费者才能提交offset。poll 的时间间隔检测能保证消费者的活跃 active

生产者，消费者【非常好的描述】

生产者

生产者可以将数据发布到所选择的topic（主题）中。生产者负责将记录分配到topic的哪一个 partition（分区）中。可以使用循环的方式来简单地实现负载均衡，也可以根据某些语义分区函数(例如：记录中的key)来完成。下面会介绍更多关于分区的使用。

消费者

消费者使用一个 消费组 名称来进行标识，发布到topic中的每条记录被分配给订阅消费组中的一个消费者实例.消费者实例可以分布在多个进程中或者多个机器上。

• 如果所有的消费者实例在同一消费组中，消息记录会负载平衡到一个消费者实例（只有一个消费者实例能消费一条消息）.

• 如果所有的消费者实例在不同的消费组中，每条消息记录会广播到所有的消费者进程（所有的消费者都能想消费同一条消息）.

如图，这个 Kafka 集群有两台 server 的，四个分区(p0-p3)和两个消费者组。消费组A有两个消费者，消费组B有四个消费者。

通常情况下，每个 topic 都会有一些消费组，**一个消费组对应一个”逻辑订阅者”**。一个消费组由许多消费者实例组成，便于扩展和容错。这就是发布和订阅的概念，只不过订阅者是一组消费者而不是单个的进程。

在Kafka中实现消费的方式是将日志中的分区划分到每一个消费者实例上，以便在任何时间，每个实例都是分区唯一的消费者。维护消费组中的消费关系由Kafka协议动态处理。如果新的实例加入组，他们将从组中其他成员处接管一些 partition 分区;如果一个实例消失，拥有的分区将被分发到剩余的实例。

Kafka 只保证分区内的记录是有序的，而不保证主题中不同分区的顺序。每个 partition 分区按照key值排序足以满足大多数应用程序的需求。但如果你需要总记录在所有记录的上面，可使用仅有一个分区的主题来实现，这意味着每个消费者组只有一个消费者进程。

kafka的保证

high-level Kafka给予以下保证:

• 生产者发送到特定topic partition 的消息将按照发送的顺序处理。 也就是说，如果记录M1和记录M2由相同的生产者发送，并先发送M1记录，那么M1的偏移比M2小，并在日志中较早出现
• 一个消费者实例按照日志中的顺序查看记录.
• 对于具有N个副本的主题，我们最多容忍N-1个服务器故障，从而保证不会丢失任何提交到日志中的记录.

如何保证数据不丢失

broker如何保证数据的不丢失

• acks=all : 所有副本都写入成功并确认。
• retries = 一个合理值。
• min.insync.replicas=2 消息至少要被写入到这么多副本才算成功。
• unclean.leader.election.enable=false 关闭unclean leader选举，即不允许非ISR中的副本被选举为leader，以避免数据丢失。

consumer如果保证数据得不丢失

enable.auto.commit=false 关闭自动提交offset。

kafka的使用

Kafka作为消息系统

传统的消息系统有两个模块: 队列 和 发布-订阅。 在队列中，消费者池从server读取数据，每条记录被池子中的一个消费者消费; 在发布订阅中，记录被广播到所有的消费者。两者均有优缺点。 队列的优点在于它允许你将处理数据的过程分给多个消费者实例，使你可以扩展处理过程。 不好的是，队列不是多订阅者模式的—一旦一个进程读取了数据，数据就会被丢弃。 而发布-订阅系统允许你广播数据到多个进程，但是无法进行扩展处理，因为每条消息都会发送给所有的订阅者。

消费组在Kafka有两层概念。在队列中，消费组允许你将处理过程分发给一系列进程(消费组中的成员)。 在发布订阅中，Kafka允许你将消息广播给多个消费组。

Kafka的优势在于每个topic都有以下特性—可以扩展处理并且允许多订阅者模式—不需要只选择其中一个.

Kafka相比于传统消息队列还具有更严格的顺序保证 ……

Kafka 设计的更好。topic中的partition是一个并行的概念。 Kafka能够为一个消费者池提供顺序保证和负载平衡，是通过将topic中的partition分配给消费者组中的消费者来实现的， 以便每个分区由消费组中的一个消费者消耗。通过这样，我们能够确保消费者是该分区的唯一读者，并按顺序消费数据。 众多分区保证了多个消费者实例间的负载均衡。但请注意，消费者组中的消费者实例个数不能超过分区的数量。

Kafka 作为存储系统

许多消息队列可以发布消息，除了消费消息之外还可以充当中间数据的存储系统。那么Kafka作为一个优秀的存储系统有什么不同呢?

数据写入Kafka后被写到磁盘，并且进行备份以便容错。直到完全备份，Kafka才让生产者认为完成写入，即使写入失败Kafka也会确保继续写入

Kafka使用磁盘结构，具有很好的扩展性—50kb和50TB的数据在server上表现一致。

可以存储大量数据，并且可通过客户端控制它读取数据的位置，您可认为Kafka是一种高性能、低延迟、具备日志存储、备份和传播功能的分布式文件系统。

Kafka用做流处理

Kafka 流处理不仅仅用来读写和存储流式数据，它最终的目的是为了能够进行实时的流处理。

在Kafka中，流处理器不断地从输入的topic获取流数据，处理数据后，再不断生产流数据到输出的topic中去。

多种消费者模式

生产消费者模式

搞清楚了Kafka的基本概念后，我们来看如何设计生产消费者模式来实现上述的“数据接入”场景。在下图中，由Producer负责接收前端上报的数据，投递到对应的topic中（这里忽略了Broker Server的细节），在Consumer端，所有对该数据感兴趣的业务都可以建立自己的group来消费数据，至于group内部开多少个worke来消费完全取决于数据量和业务的实时性要求了。

发布订阅模式

再来看“事件分发”的场景，假如我们有“收藏”、“下单”、“付款”三个事件，业务一对“收藏”和“下单”事件感兴趣，而业务二对“下单”和“付款”事件感兴趣，那么我们如何进行事件订阅？Kafka只提供了单播和广播的消息模型，无法直接进行消费对象的绑定，所以理论上Kafka是不适合做此种场景下的订阅发布模式的，如果一定要做，有这么几个方案：

• 方案一：继续使用上述生产消费者的模式，在不同的group中过滤出自己感兴趣的事件数据，然后进行处理。这种方式简单有效，缺点就是每个group都会收到很多自己不感兴趣的垃圾数据。
• 方案二：把每个事件的数据推送到不同的topic中，即以事件名称来作为topic分类，在Consumer端，建立自己的group来消费自己感兴趣的一组topic。这种方式适用于事件个数可以明确评估并且数量较少，如果事件种类很多，会导致topic的数量过多，创建过多的topic和partition则会影响到Kafka的性能，因为Kafka的每个Topic、每个分区都会对应一个物理文件，当Topic数量增加时，消息分散的落盘策略会导致磁盘IO竞争激烈成为瓶颈。

参考

官方文档，看这一篇就够了

kafka分区（partition）和和分组（group）

Kafka下的生产消费者模式与订阅发布模式