铜陵58同城做网站,自己建站的网站,华大 建设网站,高端网站建设价格要确保Kafka在使用过程中的稳定性#xff0c;需要从kafka在业务中的使用周期进行依次保障。主要可以分为#xff1a;事先预防#xff08;通过规范的使用、开发#xff0c;预防问题产生#xff09;、运行时监控#xff08;保障集群稳定#xff0c;出问题能及时发现#…要确保Kafka在使用过程中的稳定性需要从kafka在业务中的使用周期进行依次保障。主要可以分为事先预防通过规范的使用、开发预防问题产生、运行时监控保障集群稳定出问题能及时发现、故障时解决有完整的应急预案这三阶段。 另外的篇幅请参考
如何更好地使用Kafka? - 事先预防篇-CSDN博客
如何更好地使用Kafka? - 故障时解决-CSDN博客 1. 事先预防原则
事先预防即通过规范的使用、开发预防问题产生。主要包含集群/生产端/消费端的一些最佳实践、上线前测试以及一些针对紧急情况如消息积压等的临时开关功能。
Kafka调优原则 确定优化目标并且定量给出目标Kafka 常见的优化目标是吞吐量、延时、持久性和可用性 确定了目标之后需要明确优化的维度 通用性优化操作系统、JVM 等 针对性优化优化 Kafka 的 TPS、处理速度、延时等。 2. 生产端最佳实践
2.1 参数调优 使用 Java 版的 Client 使用 kafka-producer-perf-test.sh 测试你的环境 设置内存、CPU、batch 压缩 batch.size该值设置越大吞吐越大但延迟也会越大 linger.ms表示 batch 的超时时间该值越大吞吐越大、但延迟也会越大 max.in.flight.requests.per.connection默认为5表示 client 在 blocking 之前向单个连接broker发送的未确认请求的最大数超过1时将会影响数据的顺序性 compression.type压缩设置会提高吞吐量 acks数据 durability 的设置 避免大消息占用过多内存、降低broker处理速度 broker调整增加 num.replica.fetchers提升 Follower 同步 TPS避免 Broker Full GC 等 当吞吐量小于网络带宽时增加线程、提高 batch.size、增加更多 producer 实例、增加 partition 数 设置 acks-1 时如果延迟增大可以增大 num.replica.fetchersfollower 同步数据的线程数来调解 跨数据中心的传输增加 socket 缓冲区设置以及 OS tcp 缓冲区设置。 2.2 开发实践
(1) 做好Topic隔离
根据具体场景是否允许一定延迟、实时消息、定时周期任务等区分kafka topic避免挤占或阻塞实时业务消息的处理。
(2) 做好消息流控
如果下游消息消费存在瓶颈或者集群负载过高等需要在生产端或消息网关实施流量生产速率的控制或者延时/暂定消息发送等策略避免短时间内发送大量消息。
(3) 做好消息补推
手动去查询丢失的那部分数据然后将消息重新发送到mq里面把丢失的数据重新补回来。
(4) 做好消息顺序性保障
如果需要在保证Kafka在分区内严格有序的话即需要保证两个消息是有严格的先后顺序需要设置key让某类消息根据指定规则路由到同一个topic的同一个分区中能解决大部分消费顺序的问题。 但是需要避免分区内消息倾斜的问题例如按照店铺Id进行路由容易导致消息不均衡的问题。 生产端消息发送指定key确保相同key的消息发送到同一个partition。 消费端单线程消费或者写 N 个内存 queue具有相同 key 的数据都到同一个内存 queue然后对于 N 个线程每个线程分别消费一个内存 queue
(5) 适当提高消息发送效率 批量发送kafka先将消息缓存在内存中的双端队列buffer中当消息量达到batch size指定大小时进行批量发送减少了网络传输频次提高了传输效率 端到端压缩消息 将一批消息打包后进行压缩发送给 Broker 服务器后但频繁的压缩和解压也会降低性能最终还是以压缩的方式传递到消费者的手上在 Consumer 端进行解压 异步发送将生产者改造为异步的方式可以提升发送效率但是如果消息异步产生过快会导致挂起线程过多内存不足最终导致消息丢失 索引分区并行消费当一个时间相对长的任务在执行时它会占用该消息所在索引分区被锁定后面的任务不能及时派发给空闲的客户端处理若服务端如果启用索引分区并行消费的特性就可以及时的把后面的任务派发给其他的客户端去执行同时也不需要调整索引的分区数但此类消息仅适用于无需保证消息顺序关系的消息
(6) 保证消息发送可靠性 Producer如果对数据可靠性要求很高的话在发送消息的时候需要选择带有 callBack 的api进行发送并设置 acks、retries、factor等等些参数来保证Producer发送的消息不丢失。 Brokerkafka为了得到更高的性能和吞吐量将数据异步批量的存储在磁盘中并采用了批量刷盘的做法如果对数据可靠性要求很高的话可以修改为同步刷盘的方式提高消息的可靠性。 3. 消费端最佳实践
3.1 参数调优 吞吐量调整partition 数、OS page cache分配足够的内存来缓存数据 offset topic__consumer_offsetsoffsets.topic.replication.factor默认为3、offsets.retention.minutes默认为1440即 1day offset commit较慢异步 commit 或 手动 commit fetch.min.bytes 、fetch.max.wait.ms max.poll.interval.ms调用 poll() 之后延迟的最大时间超过这个时间没有调用 poll() 的话就会认为这个 consumer 挂掉了将会进行 rebalance max.poll.records当调用 poll() 之后返回最大的 record 数默认为500 session.timeout.ms Consumer Rebalancecheck timeouts、check processing times/logic、GC Issues 网络配置
3.2 开发实践
(1) 做好消息消费幂等
消息消费的幂等主要根据业务逻辑做调整。
以处理订单消息为例 由订单编号订单状态唯一的幂等key并存入redis 在处理之前首先会去查Redis是否存在该Key如果存在则说明已经处理过了直接丢掉 如果Redis没处理过则将处理过的数据插入到业务DB上再到最后把幂等Key插入到Redis上
简而言之即通过Redis做前置处理 DB唯一索引做最终保证来实现幂等性。 (2) 做好Consumer隔离
在消息量非常大的情况下实时和离线消费者同时消费一个集群离线数据繁重的磁盘 IO 操作会直接影响实时业务的实时性和集群的稳定性。
根据消费的实时性可以将消息消费者行为划分两类实时消费者和离线消费者。 实时消费者对数据实时性要求较高在实时消费的场景下Kafka 会利用系统的 page cache 缓存直接从内存转发给实时消费者热读磁盘压力为零适合广告、推荐等业务场景。 离线消费者定时周期性消费者通常是消费数分钟前或是数小时前的消息这类消息通常存储在磁盘中消费时会触发磁盘的 IO 操作冷读适合报表计算、批量计算等周期性执行的业务场景。 (3) 避免消息消费堆积 延迟处理、控制速度时间范围内分摊消息针对实时性不高的消息 生产速度大于消费速度这样可以适当增加分区增加consumer数量提升消费TPS 避免很重的消费逻辑优化consumer TPS 是否有大量DB操作 下游/外部服务接口调用超时 是否有lock操作导致线程阻塞 需要特别关注kafka异步链路中的涉及消息放大的逻辑 如果有较重的消费逻辑需要调整xx参数避免消息没消费完时消费组退出造成reblance等问题 确保consumer端没有因为异常而导致消费hang住; 如果使用的是消费者组确保没有频繁地发生rebalance 多线程消费批量拉取处理 注批量拉取处理时需注意下kafka版本spring-kafka 2.2.11.RELEASE版本以下如果配置kafka.batchListenertrue但是将消息接收的元素设置为单个元素非批量List可能会导致kafka在拉取一批消息后仅仅消费了头部的第一个消息。 (4) 避免Rebalance问题
A. 触发条件 消费者数量变化 新消费者加入、消费者下线未能及时发送心跳被“踢出”Group、消费者主动退出消费组Consumer 消费时间过长导致 消费组内订阅的主题或者主题的分区数量发生变化 消费组对应的 GroupCoorinator 节点发生变化
B. 如何避免非必要rebalance消费者下线、消费者主动退出消费组导致的reblance 需要仔细地设置session.timeout.ms决定了 Consumer 存活性的时间间隔和heartbeat.interval.ms控制发送心跳请求频率的参数 的值。 max.poll.interval.ms参数配置控制 Consumer 实际消费能力对 Rebalance 的影响限定了 Consumer 端应用程序两次调用 poll 方法的最大时间间隔。默认值是 5 分钟表示 Consumer 程序如果在 5 分钟之内无法消费完 poll 方法返回的消息那么 Consumer 会主动发起“离开组”的请求Coordinator 也会开启新一轮 Rebalance。具体可以统计下历史的时间花费把最长的时间为参考进行设置。
(5) 保证消息消费可靠性
一般情况下还是client 消费 broker 丢消息的场景比较多想client端消费数据不能丢肯定是不能使用autoCommit的所以必须是手动提交的。
Consumer自动提交的机制是根据一定的时间间隔将收到的消息进行commit。commit过程和消费消息的过程是异步的。也就是说可能存在消费过程未成功比如抛出异常commit消息已经提交了则此时消息就丢失了。 (6) 保证消息消费顺序性 不同topic乱序消息如果支付与订单生成对应不同的topic只能在consumer层面去处理了。 同一个topic乱序消息一个topic可以对应多个分区分别对应了多个consumer与“不同topic”没什么本质上的差别。可以理解为我们的服务有多个pod生产者顺序发送消息但被路由到不同分区就可能变得乱序了服务消费的就是无序的消息 同一个topic同一个分区顺序消息Kafka的消息在分区内是严格有序的例如把同一笔订单的所有消息按照生成的顺序一个个发送到同一个topic的同一个分区。
针对乱序消息
例如订单和支付分别封装了各自的消息但是消费端的业务场景需要按订单消息-支付消息的顺序依次消费消息。 宽表业务主题相关的指标、维度、属性关联在一起的一张数据库表消费消息时只更新对应的字段就好消息只会存在短暂的状态不一致问题但是状态最终是一致的。例如订单支付有自己的状态字段订单有自己的状态字段售后有自己的状态字段就不需要保证支付、订单、售后消息的有序即使消息无序也只会更新自己的状态字段不会影响到其他状态 消息补偿机制将消息与DB进行对比如果发现数据不一致再重新发送消息至主进程处理保证最终一致性 MQ队列一个中间方比如redis的队列来维护MQ的顺序 业务保证通过业务逻辑保障消费顺序
针对顺序消息
两者都是通过将消息绑定到定向的分区或者队列来保证顺序性通过增加分区或者线程来提升消费能力。
A. Consumer单线程顺序消费
生产者在发送消息时已保证消息在分区内有序一个分区对应了一个消费者保证了消息消费的顺序性。 B. Consumer多线程顺序消费具体策略在后面章节
单线程顺序消费的扩展能力很差。为了提升消费者的处理速度除了横向扩展分区数增加消费者外还可以使用多线程顺序消费。
将接收到的kafka数据进行hash取模注意如果kafka分区接受消息已经是取模的了这里一定要对id做一次hash再取模发送到不同的队列然后开启多个线程去消费对应队列里面的数据。
此外这里通过配置中心进行开关、动态扩容/缩容线程池。
(7) 处理Consumer的事务
通过事务消息可以很好的保证一些业务场景的事务逻辑不会因为网络不可用等原因出现系统之间状态不一致。
当更新任何一个服务出现故障时就抛出异常事务消息不会被提交或回滚消息服务器会回调发送端的事务查询接口确定事务状态发送端程序可以根据消息的内容对未做完的任务重新执行然后告诉消息服务器该事务的状态。 4. 集群配置最佳实践
4.1 集群配置 Broker 评估每个 Broker 的 Partition 数不应该超过2k、控制 partition 大小不要超过25GB 集群评估Broker 的数量根据以下条件配置数据保留时间、集群的流量大小 集群扩容磁盘使用率应该在 60% 以下、网络使用率应该在 75% 以下 集群监控保持负载均衡、确保 topic 的 partition 均匀分布在所有 Broker 上、确保集群的阶段没有耗尽磁盘或带宽
4.2 Topic 评估 Partition 数 Partition 数应该至少与最大 consumer group 中 consumer 线程数一致 对于使用频繁的 topic应该设置更多的 partition 控制 partition 的大小25GB 左右 考虑应用未来的增长可以使用一种机制进行自动扩容 使用带 key 的 topic partition 扩容当 partition 的数据量超过一个阈值时应该自动扩容实际上还应该考虑网络流量。
4.3 分区配置
设置多个分区在一定程度上是可以提高消费者消费的并发度但是分区数量过多时可能会带来句柄开销过大、生产端占用内存过大、可能增加端到端的延迟、影响系统可用性、故障恢复时间较长等问题。
根据吞吐量的要求设置 partition 数 假设 Producer 单 partition 的吞吐量为 P consumer 消费一个 partition 的吞吐量为 C 而要求的吞吐量为 T 那么 partition 数至少应该大于 T/P、T/c 的最大值 5. 性能调优
调优目标高吞吐量、低延时。
5.1 分层调优
自上而下分为应用程序层、框架层、JVM层和操作系统层层级越靠上调优的效果越明显。 调优类型 建议 操作系统 挂载文件系统时禁掉atime更新选择ext4或XFS文件系统swap空间的设置页缓存大小 JVM堆设置和GC收集器 将JVM 堆大小设置成 68GB建议使用 G1 收集器方便省事比 CMS 收集器的优化难度小 Broker端 保持服务器端和客户端版本一致 应用层 要频繁地创建Producer和Consumer对象实例用完及时关闭合理利用多线程来改善性能 5.2 吞吐量(TPS)调优 参数列表 Broker端 适当增加num.replica.fetchers参数值但不超过CPU核数 调优GC参数以避免经常性的Full GC Producer端 适当增加batch.size参数值比如从默认的16KB增加到512KB或1MB 适当增加linger.ms参数值比如10100 设置compression.typelz4或zstd 设置acks0或1 设置retries0 如果多线程共享同一个Producer实例则增加buffer.memory参数值 Consumer端 采用多Consumer进程或线程同时消费数据 增加fetch.min.bytes参数值比如设置成1KB或更大
5.3 延时调优 参数列表 Broker端 适当设置num.replica.fetchers值 Producer端 设置linger.ms0 不启用压缩即设置compression.typenone 设置ackes1 Consumer端 设置fetch.min.bytes1 6. 稳定性测试
kafka的稳定性测试主要在业务上线前针对Kafka实例/集群健康性、高可用性的测试。
6.1 健康性检查
(1) 检查实例查看Kafka 实例对象中拿到所有的信息例如 IP、端口等
(2) 测试可用性访问生产者和消费者测试连接。
6.2 高可用测试
A. 单节点异常测试重启Leader副本或Follower副本所在Pod
步骤 查看topic的副本信息 删除相应pod 脚本检测Kafka的可用性
预期对生产者和消费者的可用性均无影响。
B. 集群异常测试重启所有pod
步骤 删除所有pod 脚本检测Kafka的可用性
预期所有broker ready后服务正常。
