京东网站建设步骤,网页加速器手机版哪个好,30岁学网站建设,百度app首页第三章 Kafka生产问题总结及性能优化实践
1、线上环境规划 JVM参数设置
kafka 是 scala 语言开发#xff0c;运行在 JVM 上#xff0c;需要对 JVM 参数合理设置#xff0c;参看 JVM 调优专题
修改 bin/kafka-start-server.sh 中的 JVM 设置#xff0c;假设机器是 32G 内…第三章 Kafka生产问题总结及性能优化实践
1、线上环境规划 JVM参数设置
kafka 是 scala 语言开发运行在 JVM 上需要对 JVM 参数合理设置参看 JVM 调优专题
修改 bin/kafka-start-server.sh 中的 JVM 设置假设机器是 32G 内存可以如下设置
export KAFKA_HEAP_OPTS-Xmx16G -Xms16G -Xmn10G -XX:MetaspaceSize256M -XX:UseG1GC -XX:MaxGCPauseMillis50 -XX:G1HeapRegionSize16M这种大内存的情况一般都要用 G1 垃圾收集器因为年轻代内存比较大用 G1 可以设置 GC 最大停顿时间不至于一次 minor gc 就花费太长时间当然因为像 kafkarocketmqes 这些中间件写数据到磁盘会用到操作系统的 page cache所以 JVM 内存不宜分配过大需要给操作系统的缓存留出几个 G。
2、线上问题及优化
消息丢失情况 消息发送端 1、acks0 表示 producer 不需要等待任何 broker 确认收到消息的回复就可以继续发送下一条消息。性能最高但是最容易丢消息。大数据统计报表场景对性能要求很高对数据丢失不敏感的情况可以用这种。
2、acks1 至少要等待 leader 已经成功将数据写入本地log但是不需要等待所有 follower 是否成功写入。就可以继续发送下一条消息。这种情况下如果follower 没有成功备份数据而此时 leader 又挂掉则消息会丢失。
3、acks-1或all 这意味着leader需要等待所有备份(min.insync.replicas配置的备份个数)都成功写入日志这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别或跟钱打交道的场景才会使用这种配置。当然如果 min.insync.replicas 配置的是1则也可能丢消息跟acks1情况类似。 消息消费端 如果消费这边配置的是自动提交万一消费到数据还没处理完就自动提交offset了但是此时你consumer直接宕机了未处理完的数据丢失了下次也消费不到了。
消息重复消费 消息发送端 发送消息如果配置了重试机制比如网络抖动时间过长导致发送端发送超时实际broker可能已经接收到消息但发送方会重新发送消息 消息消费端 如果消费这边配置的是自动提交刚拉取了一批数据处理了一部分但还没来得及提交服务挂了下次重启又会拉取相同的一批数据重复处理
一般消费端都是要做消费幂等处理的。
消息乱序
如果发送端配置了重试机制kafka不会等之前那条消息完全发送成功才去发送下一条消息这样可能会出现发送了123条消息第一条超时了后面两条发送成功再重试发送第1条消息这时消息在broker端的顺序就是231了。
所以是否一定要配置重试要根据业务情况而定。也可以用同步发送的模式去发消息当然acks不能设置为0这样也能保证消息发送的有序。
kafka保证全链路消息顺序消费需要从发送端开始将所有有序消息发送到同一个分区然后用一个消费者去消费但是这种性能比较低可以在消费者端接收到消息后将需要保证顺序消费的几条消费发到内存队列可以搞多个一个内存队列开启一个线程顺序处理消息。
消息积压
1、线上有时因为发送方发送消息速度过快或者消费方处理消息过慢可能会导致broker积压大量未消费消息。 此种情况如果积压了上百万未消费消息需要紧急处理可以修改消费端程序让其将收到的消息快速转发到其他topic可以设置很多分区然后再启动多个消费者同时消费新主题的不同分区。
2、由于消息数据格式变动或消费者程序有bug导致消费者一直消费不成功也可能导致 broker 积压大量未消费消息。
此种情况可以将这些消费不成功的消息转发到其它队列里去(类似死信队列)后面再慢慢分析死信队列里的消息处理问题。
延时队列
延时队列存储的对象是延时消息。所谓的“延时消息”是指消息被发送以后并不想让消费者立刻获取而是等待特定的时间后消费者才能获取这个消息进行消费延时队列的使用场景有很多 比如
1、在订单系统中 一个用户下单之后通常有 30 分钟的时间进行支付如果 30 分钟之内没有支付成功那么这个订单将进行异常处理这时就可以使用延时队列来处理这些订单了。
2、订单完成 1 小时后通知用户进行评价。 实现思路 发送延时消息时先把消息按照不同的延迟时间段发送到指定的队列中topic_1stopic_5stopic_10s…topic_2h这个一般不能支持任意时间段的延时然后通过定时器进行轮训消费这些 topic查看消息是否到期如果到期就把这个消息发送到具体业务处理的 topic 中队列中消息越靠前的到期时间越早具体来说就是定时器在一次消费过程中对消息的发送时间做判断看下是否延迟到对应时间了如果到了就转发如果还没到这一次定时任务就可以提前结束了。
消息回溯
如果某段时间对已消费消息计算的结果觉得有问题可能是由于程序 bug 导致的计算错误当程序 bug 修复后这时可能需要对之前已消费的消息进行重新消费可以指定从多久之前的消息回溯消费这种可以用 consumer 的 offsetsForTimes、seek 等方法指定从某个 offset 偏移的消息开始消费参见上节课的内容。
分区数越多吞吐量越高吗
可以用 kafka 压测工具自己测试分区数不同各种情况下的吞吐量
# 往 test 里发送一百万条消息每条消息设置 1 KB
# throughput 用来进行限流控制当设定的值小于 0 时不限流当设定的值大于 0 时当发送的吞吐量大于该值时就会被阻塞一段时间
bin/kafka-producer-perf-test.sh --topic test --num-records 1000000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers192.168.65.60:9092 acks1网络上很多资料都说分区数越多吞吐量越高 但从压测结果来看分区数到达某个值时吞吐量反而开始下降实际上很多事情都会有一个临界值当超过这个临界值之后很多原本符合既定逻辑的走向又会变得不同。一般情况下分区数跟集群机器数量相当就差不多了。
当然吞吐量的数值和走势还会和磁盘、文件系统、 I/O调度策略等因素相关。
注意如果分区数设置过大比如设置10000可能会设置不成功后台会报错java.io.IOException : Too many open files。
异常中最关键的信息是“ Too many open flies”这是一种常见的 Linux 系统错误通常意味着文件描述符不足它一般发生在创建线程、创建 Socket、打开文件这些场景下 。 在 Linux 系统的默认设置下这个文件描述符的个数不是很多 通过 ulimit -n 命令可以查看一般默认是1024可以将该值增大比如ulimit -n 65535
消息传递保障
at most once消费者最多收到一次消息0–1次acks 0 可以实现。at least once消费者至少收到一次消息1–多次ack all 可以实现。exactly once消费者刚好收到一次消息at least once 加上消费者幂等性可以实现还可以用 kafka 生产者的幂等性来实现。 kafka 生产者的幂等性 因为发送端重试导致的消息重复发送问题kafka 的幂等性可以保证重复发送的消息只接收一次只需在生产者加上参数 props.put(“enable.idempotence”, true) 即可默认是 false 不开启。 具体实现原理是 kafka 每次发送消息会生成 PID 和 Sequence Number并将这两个属性一起发送给 brokerbroker 会将 PID 和 Sequence Number 跟消息绑定一起存起来下次如果生产者重发相同消息broker 会检查 PID 和 Sequence Number如果相同则不会再接收。
PID每个新的 Producer 在初始化的时候会被分配一个唯一的 PID这个PID 对用户完全是透明的。生产者如果重启则会生成新的PID。
Sequence Number对于每个 PID该 Producer 发送到每个 Partition 的数据都有对应的序列号这些序列号是从0开始单调递增的。kafka的事务
Kafka 的事务不同于 RocketmqRocketmq是保障本地事务比如数据库与 mq 消息发送的事务一致性Kafka 的事务主要是保障一次发送多条消息的事务一致性要么同时成功要么同时失败一般在 kafka 的流式计算场景用得多一点比如kafka 需要对一个 topic 里的消息做不同的流式计算处理处理完分别发到不同的 topic 里这些 topic 分别被不同的下游系统消费比如hbaseredises等这种我们肯定希望系统发送到多个 topic 的数据保持事务一致性。Kafka 要实现类似 Rocketmq 的分布式事务需要额外开发功能。
Properties props new Properties();props.put(bootstrap.servers, localhost:9092);props.put(transactional.id, my-transactional-id);ProducerString, String producer new KafkaProducer(props, new StringSerializer(), new StringSerializer());//初始化事务producer.initTransactions();try {//开启事务producer.beginTransaction();for (int i 0; i 100; i){//发到不同的主题的不同分区producer.send(new ProducerRecord(hdfs-topic, Integer.toString(i), Integer.toString(i)));producer.send(new ProducerRecord(es-topic, Integer.toString(i), Integer.toString(i)));producer.send(new ProducerRecord(redis-topic, Integer.toString(i), Integer.toString(i)));}//提交事务producer.commitTransaction();} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {// We cant recover from these exceptions, so our only option is to close the producer and exit.producer.close();} catch (KafkaException e) {// For all other exceptions, just abort the transaction and try again.//回滚事务producer.abortTransaction();}producer.close();3、 kafka高性能的原因
文章转载自： http://www.morning.bnbtp.cn.gov.cn.bnbtp.cn http://www.morning.qtbnm.cn.gov.cn.qtbnm.cn http://www.morning.fpxms.cn.gov.cn.fpxms.cn http://www.morning.lgsfb.cn.gov.cn.lgsfb.cn http://www.morning.npmpn.cn.gov.cn.npmpn.cn http://www.morning.tjwfk.cn.gov.cn.tjwfk.cn http://www.morning.hgfxg.cn.gov.cn.hgfxg.cn http://www.morning.tkxr.cn.gov.cn.tkxr.cn http://www.morning.mlyq.cn.gov.cn.mlyq.cn http://www.morning.zcnfm.cn.gov.cn.zcnfm.cn http://www.morning.pqktp.cn.gov.cn.pqktp.cn http://www.morning.qbksx.cn.gov.cn.qbksx.cn http://www.morning.zbgqt.cn.gov.cn.zbgqt.cn http://www.morning.kbyp.cn.gov.cn.kbyp.cn http://www.morning.dswtz.cn.gov.cn.dswtz.cn http://www.morning.wtsr.cn.gov.cn.wtsr.cn http://www.morning.ebpz.cn.gov.cn.ebpz.cn http://www.morning.dlwzm.cn.gov.cn.dlwzm.cn http://www.morning.qwlml.cn.gov.cn.qwlml.cn http://www.morning.dlmqn.cn.gov.cn.dlmqn.cn http://www.morning.mdgpp.cn.gov.cn.mdgpp.cn http://www.morning.pbksb.cn.gov.cn.pbksb.cn http://www.morning.qfdyt.cn.gov.cn.qfdyt.cn http://www.morning.rwzqn.cn.gov.cn.rwzqn.cn http://www.morning.swzpx.cn.gov.cn.swzpx.cn http://www.morning.lmmkf.cn.gov.cn.lmmkf.cn http://www.morning.jjzxn.cn.gov.cn.jjzxn.cn http://www.morning.dygqq.cn.gov.cn.dygqq.cn http://www.morning.kpcdc.cn.gov.cn.kpcdc.cn http://www.morning.lbhck.cn.gov.cn.lbhck.cn http://www.morning.bdqpl.cn.gov.cn.bdqpl.cn http://www.morning.zcsch.cn.gov.cn.zcsch.cn http://www.morning.qjxkx.cn.gov.cn.qjxkx.cn http://www.morning.jgmdr.cn.gov.cn.jgmdr.cn http://www.morning.qpqwd.cn.gov.cn.qpqwd.cn http://www.morning.gbcxb.cn.gov.cn.gbcxb.cn http://www.morning.bmfqg.cn.gov.cn.bmfqg.cn http://www.morning.mhrzd.cn.gov.cn.mhrzd.cn http://www.morning.ffdyy.cn.gov.cn.ffdyy.cn http://www.morning.rfhmb.cn.gov.cn.rfhmb.cn http://www.morning.brbmf.cn.gov.cn.brbmf.cn http://www.morning.jtkfm.cn.gov.cn.jtkfm.cn http://www.morning.zpzys.cn.gov.cn.zpzys.cn http://www.morning.znpyw.cn.gov.cn.znpyw.cn http://www.morning.bmzxp.cn.gov.cn.bmzxp.cn http://www.morning.gcfrt.cn.gov.cn.gcfrt.cn http://www.morning.qgjxt.cn.gov.cn.qgjxt.cn http://www.morning.rtlth.cn.gov.cn.rtlth.cn http://www.morning.fjfjm.cn.gov.cn.fjfjm.cn http://www.morning.sacxbs.cn.gov.cn.sacxbs.cn http://www.morning.wbhzr.cn.gov.cn.wbhzr.cn http://www.morning.lgtzd.cn.gov.cn.lgtzd.cn http://www.morning.sbczr.cn.gov.cn.sbczr.cn http://www.morning.bfsqz.cn.gov.cn.bfsqz.cn http://www.morning.tlfmr.cn.gov.cn.tlfmr.cn http://www.morning.tqrxm.cn.gov.cn.tqrxm.cn http://www.morning.tthmg.cn.gov.cn.tthmg.cn http://www.morning.8yitong.com.gov.cn.8yitong.com http://www.morning.rwqk.cn.gov.cn.rwqk.cn http://www.morning.rfxw.cn.gov.cn.rfxw.cn http://www.morning.xmwdt.cn.gov.cn.xmwdt.cn http://www.morning.zlrrj.cn.gov.cn.zlrrj.cn http://www.morning.qypjk.cn.gov.cn.qypjk.cn http://www.morning.ksggr.cn.gov.cn.ksggr.cn http://www.morning.xwgbr.cn.gov.cn.xwgbr.cn http://www.morning.xkwyk.cn.gov.cn.xkwyk.cn http://www.morning.plqhb.cn.gov.cn.plqhb.cn http://www.morning.zxfdq.cn.gov.cn.zxfdq.cn http://www.morning.aa1585.com.gov.cn.aa1585.com http://www.morning.smpmn.cn.gov.cn.smpmn.cn http://www.morning.nfsrs.cn.gov.cn.nfsrs.cn http://www.morning.jtszm.cn.gov.cn.jtszm.cn http://www.morning.mqxrx.cn.gov.cn.mqxrx.cn http://www.morning.qckwj.cn.gov.cn.qckwj.cn http://www.morning.czrcf.cn.gov.cn.czrcf.cn http://www.morning.gllgf.cn.gov.cn.gllgf.cn http://www.morning.xkyfq.cn.gov.cn.xkyfq.cn http://www.morning.kkwgg.cn.gov.cn.kkwgg.cn http://www.morning.drbwh.cn.gov.cn.drbwh.cn http://www.morning.jqwpw.cn.gov.cn.jqwpw.cn
