什么是品牌网站建设,平面设计培训价格,郑州 网站建设:,宣传册样式起因#xff1a;车联网项目开发#xff0c;车辆发生故障需要给三个系统推送消息#xff0c;故障上报较为频繁#xff0c;所以为了不阻塞主流程#xff0c;采用了使用kafka。消费方负责推送并保存推送记录#xff0c;但在一次压测中发现#xff0c;实际只发生了10次故障车联网项目开发车辆发生故障需要给三个系统推送消息故障上报较为频繁所以为了不阻塞主流程采用了使用kafka。消费方负责推送并保存推送记录但在一次压测中发现实际只发生了10次故障但是推送记录却有30多条。 问题排查发现是因为其中一个系统宕机导致往这个系统推送消息时一直连接超时导致每条消息的推送时长被拉长。而且kafka消息拉取参数max-poll-records设置了500意味着一次会批量拉取500条消息到本地处理而max.poll.interval.ms参数默认是5分钟当500条消息处理时长超过5分钟后就会认为消费者死掉了触发再均衡导致同一个消息被重复消费。 解决 主要是提高消费者的处理速度避免不必要的Rebalance。主要采用2种措施
减少每次拉去消息数max-poll-records从500降到20拉取到消息之后异步处理创建线程池对推送消息的部分利用多线程处理 常见配置
fetch.min.byte配置Consumer一次拉取请求中能从Kafka中拉取的最小数据量默认为1B如果小于这个参数配置的值就需要进行等待直到数据量满足这个参数的配置大小。调大可以提交吞吐量但也会造成延迟
fetch.max.bytes一次拉取数据的最大数据量默认为52428800B也就是50M但是如果设置的值过小甚至小于每条消息的值实际上也是能消费成功的
fetch.wait.max.ms若是不满足fetch.min.bytes时等待消费端请求的最长等待时间默认是500ms
max.poll.records单次poll调用返回的最大消息记录数如果处理逻辑很轻量可以适当提高该值。一次从kafka中poll出来的数据条数,max.poll.records条数据需要在在session.timeout.ms这个时间内处理完默认值为500
consumer.poll(100) 100 毫秒是一个超时时间一旦拿到足够多的数据fetch.min.bytes 参数设置consumer.poll(100)会立即返回 ConsumerRecordsString, String records。如果没有拿到足够多的数据会阻塞100ms但不会超过100ms就会返回
max.poll.interval.ms两次拉取消息的间隔默认5分钟通过消费组管理消费者时该配置指定拉取消息线程最长空闲时间若超过这个时间间隔没有发起poll操作则消费组认为该消费者已离开了消费组将进行再均衡操作将分区分配给组内其他消费者成员
若超过这个时间则报如下异常
org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has alreadyrebalanced and assigned the partitions to another member. This means that the time between subsequent calls
to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is
spending too much time message processing. You can address this either by increasing the session timeout or byreducing the maximum size of batches returned in poll() with max.poll.records. 即无法完成提交因为组已经重新平衡并将分区分配给另一个成员。这意味着对poll()的后续调用之间的时间比配置的max.poll.interval.ms长这通常意味着poll循环花费了太多的时间来处理消息。
可以通过增加max.poll.interval.ms来解决这个问题也可以通过减少在poll()中使用max.poll.records返回的批的最大大小来解决这个问题。
max.partition.fetch.bytes该属性指定了服务器从每个分区返回给消费者的最大字节数默认为 1MB。
session.timeout.ms消费者在被认为死亡之前可以与服务器断开连接的时间默认是 3s将触发再均衡操作。
对于每一个Consumer GroupKafka集群为其从Broker集群中选择一个Broker作为其Coordinator。Coordinator主要做两件事 维持Group成员的组成。这包括加入新的成员检测成员的存活性清除不再存活的成员。 协调Group成员的行为。
poll机制 每次poll的消息处理完成之后再进行下一次poll是同步操作 每次poll之前检查是否可以进行位移提交如果可以那么就会提交上一次轮询的位移 每次poll时consumer都将尝试使用上次消费的offset作为起始offset然后依次拉取消息 poll(long timeout)timeout指等待轮询缓冲区的数据所花费的时间单位是毫秒
