ps做网站教程,三亚网站建设哪家专业,国家企业信用信息,如何看网站是用什么语言做的1、消息丢失情况
消息发送端#xff1a; #xff08;1#xff09;acks0#xff1a; 表示producer不需要等待任何broker确认收到消息的回复#xff0c;就可以继续发送下一条消息。性能最高#xff0c;但是最容易丢消息。大数据统计报表场景#xff0c;对性能要求很高 1acks0 表示producer不需要等待任何broker确认收到消息的回复就可以继续发送下一条消息。性能最高但是最容易丢消息。大数据统计报表场景对性能要求很高对数据丢失不敏感的情况可以用这种。 2acks1 至少要等待leader已经成功将数据写入本地log但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。这种情况下如果follower没有成功备份数据而此时leader又挂掉则消息会丢失。 3acks-1或all 这意味着leader需要等待所有备份(min.insync.replicas配置的备份个数)都成功写入日志这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别或跟钱打交道的场景才会使用这种配置。当然如果min.insync.replicas配置的是1则也可能丢消息跟acks1情况类似。
消息消费端 如果消费这边配置的是自动提交万一消费到数据还没处理完就自动提交offset了但是此时你consumer直接宕机了未处理完的数据丢失了下次也消费不到了。
2、消息重复消费
消息发送端 发送消息如果配置了重试机制比如网络抖动时间过长导致发送端发送超时实际broker可能已经接收到消息但发送方会重新发送消息。
消息消费端 如果消费这边配置的是自动提交刚拉取了一批数据处理了一部分但还没来得及提交服务挂了下次重启又会拉取相同的一批数据重复处理。 一般消费端都是要做消费幂等处理的。
3、消息乱序
如果发送端配置了重试机制kafka不会等之前那条消息完全发送成功才去发送下一条消息这样可能会出现发送了123条消息第一条超时了后面两条发送成功再重试发送第1条消息这时消息在broker端的顺序就是231了。所以是否一定要配置重试要根据业务情况而定。也可以用同步发送的模式去发消息当然acks不能设置为0这样也能保证消息从发送端到消费端全链路有序。 kafka保证全链路消息顺序消费需要从发送端开始将所有有序消息发送到同一个分区然后用一个消费者去消费但是这种性能比较低可以在消费者端接收到消息后将需要保证顺序消费的几条消费发到内存队列(可以搞多个)一个内存队列开启一个线程顺序处理消息。
4、消息积压
1线上有时因为发送方发送消息速度过快或者消费方处理消息过慢可能会导致broker积压大量未消费消息。此种情况如果积压了上百万未消费消息需要紧急处理可以修改消费端程序让其将收到的消息快速转发到其他topic(可以设置很多分区)然后再启动多个消费者同时消费新主题的不同分区。 2由于消息数据格式变动或消费者程序有bug导致消费者一直消费不成功也可能导致broker积压大量未消费消息。 此种情况可以将这些消费不成功的消息转发到其它队列里去(类似死信队列)后面再慢慢分析死信队列里的消息处理问题。
5、延时队列
延时队列存储的对象是延时消息。所谓的“延时消息”是指消息被发送以后并不想让消费者立刻获取而是等待特定的时间后消费者才能获取这个消息进行消费延时队列的使用场景有很多 比如 1在订单系统中 一个用户下单之后通常有 30 分钟的时间进行支付如果 30 分钟之内没有支付成功那么这个订单将进行异常处理这时就可以使用延时队列来处理这些订单了。 2订单完成1小时后通知用户进行评价。
实现思路发送延时消息时先把消息按照不同的延迟时间段发送到指定的队列中topic_1stopic_5stopic_10s…topic_2h这个一般不能支持任意时间段的延时然后通过定时器进行轮训消费这些topic查看消息是否到期如果到期就把这个消息发送到具体业务处理的topic中队列中消息越靠前的到期时间越早具体来说就是定时器在一次消费过程中对消息的发送时间做判断看下是否延迟到对应时间了如果到了就转发如果还没到这一次定时任务就可以提前结束了。
6、消息回溯
如果某段时间对已消费消息计算的结果觉得有问题可能是由于程序bug导致的计算错误当程序bug修复后这时可能需要对之前已消费的消息重新消费可以指定从多久之前的消息回溯消费这种可以用consumer的offsetsForTimes、seek等方法指定从某个offset偏移的消息开始消费参见上节课的内容。
7、分区数越多吞吐量越高吗
可以用kafka压测工具自己测试分区数不同各种情况下的吞吐量
# 往test里发送一百万消息每条设置1KB
# 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
8、消息传递保障
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开始单调递增的。9、kafka的事务
Kafka的事务不同于RocketmqRocketmq是保障本地事务(比如数据库)与mq消息发送的事务一致性Kafka的事务主要是保障一次发送多条消息的事务一致性(要么同时成功要么同时失败)一般在kafka的流式计算场景用得多一点比如kafka需要对一个topic里的消息做不同的流式计算处理处理完分别发到不同的topic里这些topic分别被不同的下游系统消费(比如hbaseredises等)这种我们肯定希望系统发送到多个topic的数据保持事务一致性。Kafka要实现类似Rocketmq的分布式事务需要额外开发功能。 kafka的事务处理可以参考官方文档
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();10、kafka高性能的原因
磁盘顺序读写kafka消息不能修改以及不会从文件中间删除保证了磁盘顺序读kafka的消息写入文件都是追加在文件末尾 不会写入文件中的某个位置(随机写)保证了磁盘顺序写。 数据传输的零拷贝。 读写数据的批量batch处理以及压缩传输。
