菠菜网站的代理怎么做,网站建设 博贤科技,网站开发 面试 适当吹牛,wordpress 悬浮栏消息消费顺序问题
使用消息队列的过程中经常有业务场景需要严格保证消息的消费顺序#xff0c;比如我们同时发了 2 个消息#xff0c;这 2 个消息对应的操作分别对应的数据库操作是#xff1a;
用户等级升级。根据用户等级下的订单价格
假如这两条消息的消费顺序不一样造…消息消费顺序问题
使用消息队列的过程中经常有业务场景需要严格保证消息的消费顺序比如我们同时发了 2 个消息这 2 个消息对应的操作分别对应的数据库操作是
用户等级升级。根据用户等级下的订单价格
假如这两条消息的消费顺序不一样造成的最终结果就会截然不同。我们知道 Kafka 中 Partition(分区)是真正保存消息的地方我们发送的消息都被放在了这里。而我们的 Partition(分区) 又存在于 Topic(主题) 这个概念中并且我们可以给特定 Topic 指定多个 Partition。每次添加消息到 Partition(分区) 的时候都会采用尾加法。
Kafka 只能为我们保证 Partition(分区) 中的消息有序。消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量offset。 为什么同一个partition的消息是有序的? 因为当生产者向某个partition发送消息时消息会被追加到该ppartition的日志文件(log)中并且被分配一个唯一的offset文件的读写是有顺序的。而消费者在从该分区消费消息时会从该分区的最早offset开始逐个读取 消息保证了消息的顺序性。 如何解决
Kafka 通过偏移量offset来保证消息在分区内的顺序性。所以我们就有一种很简单的保证消息消费顺序的方法 1 个 Topic 只对应一个 Partition。这样当然可以解决问题但是破坏了 Kafka 的设计初衷。
Kafka 中发送 1 条消息的时候可以指定 topic, partition, key,data数据 4 个参数。如果你发送消息的时候指定了 Partition 的话所有消息都会被发送到指定的 Partition。并且同一个 key 的消息可以保证只发送到同一个 partition这个我们可以采用表/对象的 id 来作为 key 。
总结一下对于如何保证 Kafka 中消息消费的顺序有了下面两种方法
1 个 Topic 只对应一个 Partition。推荐发送消息的时候指定 key/Partition。
如何发到同一个partition
当我们发送消息的时候如果key为null那么Kafka默认采用Roiund-robin策略也就是轮转实现类是DefaultPartitioner。那么如果想要指定他发送到某个partition的话有以下三个方式:
指定partitionID我们可以在发送消息的时候可以直接在ProducerRecord中指定partition指定key在没有指定Partition(null值)时如果有 KeyKafka将依据Key做hash来计算出一个Partition编号来。如果key相同那么也能分到同一个partition中但这个方案如果后续要增加删除partition就会出现短暂的乱序。自定义Partitioner可以实现自己的分区器(Partitioner)来指定消息发送到特定的分区。我们需要创建一个类实现Partitioner接口并且重写partition()方法。
消息丢失问题
生产者丢失问题调优
生产者(Producer) 调用send方法发送消息之后消息可能因为网络问题并没有发送过去。所以我们不能默认在调用send方法发送消息之后消息发送成功了。为了确定消息是发送成功我们要判断消息发送的结果。
但是要注意的是 Kafka 生产者(Producer) 使用 send 方法发送消息实际上是异步的操作我们可以通过 get()方法获取调用结果但是这样也让它变为了同步操作不推荐这么做可以采用为其添加回调函数的形式。
如果消息发送失败的话我们检查失败的原因之后重新发送即可另外这里推荐为 Producer 的retries重试次数设置一个比较合理的值一般是 3 但是为了保证消息不丢失的话一般会设置比较大一点。设置完成之后当出现网络问题之后能够自动重试消息发送避免消息丢失。另外建议还要设置重试间隔因为间隔太小的话重试的效果就不明显了网络波动一次你 3 次一下子就重试完了。
同时,我们也可以通过给producer设置一些参数来提升发送成功率:
retries3 //生产端的重试次数retry.backoff.ms 300 //消息发送超时或失败后,间隔的重试时间acks0:表示Producer请求立即返回不需要等待Leader的任何可确认。这种方案有最高的吞吐率但是不保证消息是否真的发送成功。acks-1:表示分区Leader必须等待消息被成功写入到所有的ISR副本(同步副本)中才认为Producer请求成功。这种方案提供最高的消息持久性保证但是理论上吞吐率也是最差的。acks1:表示Leader必须应答此Producer请求并写入消息到本地日志之后Producer请求被认为成功。如果此时Leader副本应答请求之后挂掉了消息会丢失。这个方案提供了不错的持久性保证和吞吐。
消费者丢失问题调优
我们知道消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量offset。偏移量offset)表示 Consumer 当前消费到的 Partition(分区)的所在的位置。Kafka 通过偏移量offset可以保证消息在分区内的顺序性。
当消费者拉取到了分区的某个消息之后消费者会自动提交了 offset。自动提交的话会有一个问题试想一下当消费者刚拿到这个消息准备进行真正消费的时候突然挂掉了消息实际上并没有被消费但是 offset 却被自动提交了。
解决办法也比较粗暴我们手动关闭自动提交 offset每次在真正消费完消息之后再自己手动提交 offset 。 但是细心的朋友一定会发现这样会带来消息被重新消费的问题。比如你刚刚消费完消息之后还没提交 offset结果自己挂掉了那么这个消息理论上就会被消费两次。
Broker丢消息问题调优
Kafka 为分区Partition引入了多副本Replica机制。分区Partition中的多个副本之间会有一个叫做 leader 的家伙其他副本称为 follower。我们发送的消息会被发送到 leader 副本然后 follower 副本才能从 leader 副本中拉取消息进行同步。
生产者和消费者只与 leader 副本交互。你可以理解为其他副本只是 leader 副本的拷贝它们的存在只是为了保证消息存储的安全性。
试想一种情况假如 leader 副本所在的 broker 突然挂掉那么就要从 follower 副本重新选出一个 leader 但是 leader 的数据还有一些没有被 follower 副本的同步的话就会造成消息丢失。
设置 acks all解决办法就是我们设置 acks all。acks 是 Kafka 生产者(Producer) 很重要的一个参数。
acks 的默认值即为 1代表我们的消息被 leader 副本接收之后就算被成功发送。当我们配置 acks all 表示只有所有 ISR 列表的副本全部收到消息时生产者才会接收到来自服务器的响应. 这种模式是最高级别的也是最安全的可以确保不止一个 Broker 接收到了消息.该模式的延迟会很高。
设置 replication.factor 3
为了保证 leader 副本能有 follower 副本能同步消息我们一般会为 topic 设置 replication.factor 3。这样就可以保证每个 分区(partition) 至少有 3 个副本。虽然造成了数据冗余但是带来了数据的安全性。 设置 min.insync.replicas 1 一般情况下我们还需要设置 min.insync.replicas 1 这样配置代表消息至少要被写入到 2 个副本才算是被成功发送。min.insync.replicas 的默认值为 1 在实际生产中应尽量避免默认值 1。但是为了保证整个 Kafka 服务的高可用性你需要确保 replication.factor min.insync.replicas 。为什么呢设想一下假如两者相等的话只要是有一个副本挂掉整个分区就无法正常工作了。这明显违反高可用性一般推荐设置成 replication.factor min.insync.replicas 1。 设置 unclean.leader.election.enable false Kafka 0.11.0.0 版本开始 unclean.leader.election.enable 参数的默认值由原来的 true 改为 false 我们最开始也说了我们发送的消息会被发送到 leader 副本然后 follower 副本才能从 leader 副本中拉取消息进行同步。多个 follower 副本之间的消息同步情况不一样当我们配置了 unclean.leader.election.enable false 的话当 leader 副本发生故障时就不会从 follower 副本中和 leader 同步程度达不到要求的副本中选择出 leader 这样降低了消息丢失的可能性。
为何无法保证100%
Kafka提供的Producer和Consumer之间的消息传递保证语义有有三种所谓消息传递语义其实就是Kafka的消息交付可靠保障主要有以下三种:
At most once 消息可能会丢,但绝不会重复传递At least once 消息绝不会丢,但可能会重复传递Exactly once 每条消息只会被精确地传递一次。既不会多也不会少
目前Kafka默认提供的交付可靠性保障是第二种即At least once但是,其实依靠Kafka自身,是没有办法100%保证可靠性的。
介绍了很多调优手段但本质都是调优无法100%的保证消息一定发生成功、消费成功。
生产者问题
Kafka允许生产者以异步方式发送消息这意味着生产者在发送送消息后不会等待确认。当然我们可以注册一个回调等待消息的成功回调。
但是如果生产者在发送消息之后Kafka的集群发生故障或崩溃而消息尚未被完全写入Kafka的日志中那么这些消息可能会丢失。虽然后续有可能会重试但是如果重试也最终失败了呢如果这个过程中刚好生产者也崩溃了呢那就可能会导致没有人知道这个消息失败了就导致不会重试了。
消费者来说比较简单只要保证在消息成功时才提交偏移量量就行了这样就不会导致消息丢失了。
Broker问题
Kafka使用日志来做消息的持久化的日志文件是存储在磁盘之上的但是如果Broker在消息尚未完全写入日志之前崩溃那么这些消息可能会丢失了。
而且操作系统在写磁盘之前会先把数据写入Page Cache中然后再由操作系统中自己决定什么时候同步到磁盘当中而在这个过程中如果还没来得及同步到磁盘中就直接接宕机了那这个消息也就丢了。
当然也可以通过配置log.flush.interval.messages1来实现类似于同步刷盘的功能但是又回到了前面说的情况,还没来得及做持久化就宕机了。
即使Kafka中引入了副本机制来提升消息的可靠性但是如果发生同步延迟还没来及的同步主副本就挂掉了那么消息就可能会发生丢失。
这几种情况只从Broker的角度分析Broker自身是没办法保证消息不丢失的但是如果配合Producer再配合 request.required.acks-1 这种ACK策略可以确保消息持久化成功之后才会ACK给Producer那么如果我们的Producer在一定时间段内没有收到ACK是可以重新发送的。
但是这种重新发送就又回到了我们前面介绍生产者的时候的问题生产者也有可能挂重新发送次数也是有上限的最终发送失败还是会导致消息最终丢失。
所以只靠Kafka自己其实是没有办法保证极端情况下的消息100%不丢失的。
但是我们也可以在做一些机制来保证比如引入分布式享事务或者引入本地消息表等保证在Kafka Broker没有保存消息成功时可以重新投递消息。这样才行类似RocketMQ的实现。
