kafka批量消费（kafka批量消费原理）

本篇文章给大家谈谈kafka批量消费，以及kafka批量消费原理对应的知识点，希望对各位有所帮助，不要忘了收藏本站喔。

本文目录一览：

• 1、kafka批量发送消息实验
• 2、springboot使用kafka并发批量消费数据 （注解）
• 3、Kafka partition的数量问题
• 4、大数据Kafka有哪些优势能力呢？
• 5、多个消费者要消费kafka相同数据怎么办

kafka批量发送消息实验

批量发送：

配置

生产者

设置批量发送森枯且设置时延，当某此档洞个topic的消息超过batch-size，会把accumulator的消息全部发出去，即其他topic的也跟着一起发出去。

dosend方法详蠢清解：

[img]

springboot使用kafka并发批量消费数据 （注解）

提示：下面尘芦陵属于基础配哗或置，其他配置可以根据业务需求添加

技术有限派戚，欢迎大家更正问题，希望一起提升水准。谢谢O(∩_∩)O！

Kafka partition的数量问题

kafka的每个topic都可以创建多个partition，partition的数量无上限，并不会像replica一样受限于broker的数量，因此partition的数量可以随心所欲的设置。那确定partition的数量就需要思考一些权衡因素。

越多的partition可以提供更高的吞吐量

在kafka中，单个partition是kafka并行操作的最小单元。每个partition可以独立接收推送的消息以及被consumer消费，相当于topic的一个子通道，partition和topic的关系就像高速公路的车道和高速公路的关系一样，起始点和终点相同，每个车道都可以独立实现运输，不同的是kafka中不存在车辆变道的说法，入口时选择的车道需要从一而终。而kafka的吞吐量显而易见，在资源足够的情况下，partition越多速度越快。

这里提到的资源充足解释一下，假设我现在一个partition的最大传输速度为p，目前kafka集群共有三个broker，每个broker的资源足够支撑三个partition最大速度传输，那我的集群最大传输速度为3*3*p=9p，假设在不增加资源的情况下将partition增加到18个，每个partition只能以p/2的速度传输数据，因此传输速度上限还是9p，并不能再提升，因此吞吐量的设计需要考虑broker的资源上限。当然，kafka跟其他集群一样，可以横向扩展，再增加三个相同资源的broker，那传输速度即可达到18p。

越多的分区需要打开更多的文件句柄

在kafka的broker中，每个分区都会对照着文件系统的一个目录。

在kafka的数据日志文件目录中，每个日志数据段都会分配两个文件，一个索引文件和一个数据文件。因此，随着partition的增多，需要的文件句柄数急剧增加，必要时需要调整操作系统允许打开的文件句柄数。

更多的分区会导致端对端的延迟

kafka端对端的延迟为producer端发布消息到consumer端消费消息所需的时间，即consumer接收消息的时间减去produce发布消息的时顷稿间。kafka在消息正确接收后才会暴露给消费者，即在保证in-sync副本复制成功之后才会暴露，瓶颈则来自于此。在一个broker上的副本从其他broker的leader上局数复制数据的时候只会开启一个线程，假设partition数量为n，每个副本同步的时间为1ms，那in-sync操作完成所需的时间即n*1ms，若n为10000，则需要10秒才能返回同步状态，数据才能暴露给消费者，这就导致了较大的端对端的延迟。

越多的partition意味着需要更多的内存

在新版本的kafka中可以支持批量提交和批量消费，而设置了批量提交和批量消费后，每个partition都会需要一定的内存空间。假设为100k，当partition为100时，producer端和consumer端都需要10M的内存；当partition为100000时，producer端和consumer端则都需要10G内存。无限的partition数量很快就会占据大量的内存，造成性能瓶颈。

越多的partition会导致更长时间的恢复期

kafka通过多副本复制技术，实现kafka的高可用性和稳定性。每个partition都会有多雀腊孝个副本存在于多个broker中，其中一个副本为leader，其余的为follower。当kafka集群其中一个broker出现故障时，在这个broker上的leader会需要在其他broker上重新选择一个副本启动为leader，这个过程由kafka controller来完成，主要是从Zookeeper读取和修改受影响partition的一些元数据信息。

通常情况下，当一个broker有计划的停机上，该broker上的partition leader会在broker停机前有次序的一一移走，假设移走一个需要1ms，10个partition leader则需要10ms，这影响很小，并且在移动其中一个leader的时候，其他九个leader是可用的，因此实际上每个partition leader的不可用时间为1ms。但是在宕机情况下，所有的10个partition

leader同时无法使用，需要依次移走，最长的leader则需要10ms的不可用时间窗口，平均不可用时间窗口为5.5ms，假设有10000个leader在此宕机的broker上，平均的不可用时间窗口则为5.5s。

更极端的情况是，当时的broker是kafka controller所在的节点，那需要等待新的kafka leader节点在投票中产生并启用，之后新启动的kafka leader还需要从zookeeper中读取每一个partition的元数据信息用于初始化数据。在这之前partition leader的迁移一直处于等待状态。

总结

通常情况下，越多的partition会带来越高的吞吐量，但是同时也会给broker节点带来相应的性能损耗和潜在风险，虽然这些影响很小，但不可忽略，因此需要根据自身broker节点的实际情况来设置partition的数量以及replica的数量。

大数据Kafka有哪些优势能力呢？

Kafka的高吞吐能力、缓存机制能有效的解决高峰流量冲击问题。实践表明，在未将kafka引入系统前，当互联网关发送的数据量较大时，往往会挂起关系数据库，数据常常丢失。在引入kafka后，更新程序能够结合能力自主处理消息，不会引起数据丢失，关系型数据库的压力腊哗码波动不会发生过于显著的变化，不会出现数据库挂起锁死现象。

依靠kafka的订阅分发机制，实现了一次发布，各分支依据需求自主订阅的功能。避免了各分支机构直接向数据中心请求数据，或者数据中心依次批量向分支机构传输数据以致实时性不足的情况。kafka提高了实时性，减轻了数据中心的压力，提高了效率。为了帮助大家让学习变得轻松、高效，给大家免费分享一大批资料，帮助大家在成为大数据工轮哪程师，乃至架构师的路上披荆斩棘。在这里给大家推荐一个大数据学习交流圈：658558542 欢迎大家进群交流讨论，学习交流，共同进步。

当真正开始学习的时候难免不知道芦旁从哪入手，导致效率低下影响继续学习的信心。

但最重要的是不知道哪些技术需要重点掌握，学习时频繁踩坑，最终浪费大量时间，所以有有效资源还是很有必要的。

消费者是以consumer group消费者组的方式工作，由一个或者多个消费者组成一个组，共同消费一个topic。每个分区在同一时间只能由group中的一个消费者读取，但是多个group可以同时消费这个partition。在图中，有一个由三个消费者组成的group，有一个消费者读取主题中的两个分区，另外两个分别读取一个分区。某个消费者读取某个分区，也可以叫做某个消费者是某个分区的拥有者。

在这种情况下，消费者可以通过水平扩展的方式同时读取大量的消息。另外，如果一个消费者失败了，那么其他的group成员会自动负载均衡读取之前失败的消费者读取的分区。

消费方式

consumer采用pull（拉）模式从broker中读取数据。

push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据consumer的消费能力以适当的速率消费消息。

对于Kafka而言，pull模式更合适，它可简化broker的设计，consumer可自主控制消费消息的速率，同时consumer可以自己控制消费方式——即可批量消费也可逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义。

pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直等待数据到达。为了避免这种情况，我们在我们的拉请求中有参数，允许消费者请求在等待数据到达的“长轮询”中进行阻塞（并且可选地等待到给定的字节数，以确保大的传输大小）。

消费者组的偏移量等信息存储在zookeeper中的consumers节点中。

6.1 Kafka Producer 压力测试

record-size 是一条信息有多大，单位是字节。

num-records 是总共发送多少条信息。

throughput 是每秒多少条信息，设成-1，表示不限流，可测出生产者最大吞吐量。

多个消费者要消费kafka相同数据怎么办

在Kafak中国社区的qq群中，这个问题被提及的比例是相当高嫌斗的，这也是Kafka用户最常碰到的问题之一。本文结合Kafka源码试图对该问题相关的因素进行探讨。希望对大家有所帮助。怎么确定分区数？“我应该选择几个分区？”——如果你在Kafka中国社区的群里，这样的问题你会经常碰到的。不过有些遗憾的是，我们似乎并没有很权威的答案能够解答这样的问题。其实这也不奇怪，毕竟这样的问题通常都是没有固定答案的。Kafka官网上标榜自己是"high-throughput distributed messaging system"，即一个高吞吐量的分布式消息引擎。那么怎么达到高吞吐量呢？Kafka在底层摒弃了Java堆缓存机制，采用了操作系统级别的页缓存，同时将随机写操作改为顺序写，再结合Zero-Copy的特性极大地改善了IO性能。但是，这只是一个方面，毕竟单机优化的能力是有上限的。如何通过水平扩展甚至是线性扩展来进一步提升吞吐量呢？ Kafka就是使用了分区partition，通过将topic的消息打散到多个分区并分布保存在不同的broker上实现了消息处理不管是producer还是consumer的高吞吐量。Kafka的生产者和消费者都可以多线程地并行操作，而每个线程处理的是一个分区的数据。因此分区实际上是调优Kafka并行度的最小单元。对于producer而言，它实际上是用多个线程并发地向不同分区所在的broker发起Socket连接同时给这些分区发送消息；而consumer呢，同一个消费组内的所有consumer线程都被指定topic的某一个分区进行消费具体如何确定consumer线程数目我们后面会详细说明。所以说，如果一个topic分区越多，理论上整个集群所能达到的吞吐量就越大。但分区是否越多越好呢？显然也不是，因为每个分区都有自己的开销：一、客户端/服升绝务器端需要使用的内存就越多先说说客户端的情况。Kafka 082之后推出了Java版的全新的producer，这个producer有个参数batchsize，默认是16KB。它会为每个分区缓存消息，一旦满了就打包将消息批量发出。看上去这是个能够提升性能的设计。不过很显然，因为这个参数是分区级别的，如果分区数越多，这部分缓存所需的内存占用也会更多。假设你有10000个分区，按照默认设置，这部分缓存需要占用约157MB的内存。而consumer端呢？我们抛开获取数据所需的内存不说，只说线程的开销。如果还是假设有10000个分区，同时consumer线程数要匹配分区数大部分情况下是最佳的消费吞吐量配置的话，那么在consumer client就要创建10000个线程，也需要创建大约10000个Socket去获取分区数据。这里面的线程切换的开销本吵者姿身已经不容小觑了。服务器端的开销也不小，如果阅读Kafka源码的话可以发现，服务器端的很多组件都在内存中维护了分区级别的缓存，比如controller，FetcherManager等，因此分区数越多，这种缓存的成本越久越大。二、文件句柄的开销每个分区在底层文件系统都有属于自己的一个目录。该目录下通常会有两个文件： base_offsetlog和base_offsetindex。Kafak的controller和ReplicaManager会为每个broker都保存这两个文件句柄file handler。很明显，如果分区数越多，所需要保持打开状态的文件句柄数也就越多，最终可能会突破你的ulimit -n的限制。三、降低高可用性Kafka通过副本replica机制来保证高可用。具体做法就是为每个分区保存若干个副本replica_factor指定副本数。每个副本保存在不同的broker上。期中的一个副本充当leader 副本，负责处理producer和consumer请求。其他副本充当follower角色，由Kafka controller负责保证与leader的同步。如果leader所在的broker挂掉了，contorller会检测到然后在zookeeper的帮助下重选出新的leader——这中间会有短暂的不可用时间窗口，虽然大部分情况下可能只是几毫秒级别。但如果你有10000个分区，10个broker，也就是说平均每个broker上有1000个分区。此时这个broker挂掉了，那么zookeeper和controller需要立即对这1000个分区进行leader选举。比起很少的分区leader选举而言，这必然要花更长的时间，并且通常不是线性累加的。如果这个broker还同时是controller情况就更糟了。说了这么多“废话”，很多人肯定已经不耐烦了。那你说到底要怎么确定分区数呢？答案就是：视情况而定。基本上你还是需要通过一系列实验和测试来确定。当然测试的依据应该是吞吐量。虽然LinkedIn这篇文章做了Kafka的基准测试，但它的结果其实对你意义不大，因为不同的硬件、软件、负载情况测试出来的结果必然不一样。我经常碰到的问题类似于，官网说每秒能到10MB，为什么我的producer每秒才1MB？ —— 且不说硬件条件，最后发现他使用的消息体有1KB，而官网的基准测试是用100B测出来的，因此根本没有可比性。不过你依然可以遵循一定的步骤来尝试确定分区数：创建一个只有1个分区的topic，然后测试这个topic的producer吞吐量和consumer吞吐量。假设它们的值分别是Tp和Tc，单位可以是MB/s。然后假设总的目标吞吐量是Tt，那么分区数 = Tt / maxTp, TcTp表示producer的吞吐量。测试producer通常是很容易的，因为它的逻辑非常简单，就是直接发送消息到Kafka就好了。Tc表示consumer的吞吐量。测试Tc通常与应用的关系更大， 因为Tc的值取决于你拿到消息之后执行什么操作，因此Tc的测试通常也要麻烦一些。另外，Kafka并不能真正地做到线性扩展其实任何系统都不能，所以你在规划你的分区数的时候最好多规划一下，这样未来扩展时候也更加方便。消息-分区的分配默认情况下，Kafka根据传递消息的key来进行分区的分配，即hashkey

关于kafka批量消费和kafka批量消费原理的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。