searchusermenu
  • 发布文章
  • 消息中心
点赞
收藏
评论
分享
原创

kafka如何保证消息的顺序性

2024-08-29 09:42:16
38
0

前提

生产者给kakfa投递的消息,在同一个topic下不同的Partition中,不同的Partition之间的消息是无法保证顺序的。但有些场景,业务需要处理顺序消息。怎么保证kakfa消息的顺序性,以致最终消费者能顺序消费消息呢

消息有序性分类

  • 全局有序:一个Topic下的所有消息都需要按照生产顺序消费。
  • 局部有序:一个Topic下的消息,只需要满足同一业务字段的要按照生产顺序消费。例如:Topic消息是订单的流水表,包含订单orderId,业务要求同一个orderId的消息需要按照生产顺序进行消费。

全局有序

由于Kafka的一个Topic可以分为了多个Partition,Producer发送消息的时候,是分散在不同 Partition的。当Producer按顺序发消息给Broker,但进入Kafka之后,这些消息就不一定进到哪个Partition,会导致顺序是乱的。因此要满足全局有序,需要1个Topic只能对应1个Partition。而且对应的consumer也要使用单线程或者保证消费顺序的线程模型.

局部有序

要满足局部有序,只需要在发消息的时候指定Partition Key,Kafka对其进行Hash计算,根据计算结果决定放入哪个Partition。这样Partition Key相同的消息会放在同一个Partition。确保将相关的消息发送到同一个分区。Kafka 保证在单个分区内消息的顺序。因此,如果消息的顺序性对业务非常重要,可以将相关的消息通过相同的分区键(如用户 ID 或订单 ID)发送至同一分区。此时,Partition的数量仍然可以设置多个,提升Topic的整体吞吐量。

 

提升消费效率

在不增加partition数量的情况下想提高消费速度,可以考虑以上局部有序的场景下再次hash唯一标识(例如订单orderId)到不同的线程上,多个消费者线程并发处理消息(依旧可以保证局部有序)。说到底还是一个实例线程消费一个partition消息

消息重试对顺序消息的影响

对于一个有着先后顺序的消息A、B,正常情况下应该是A先发送完成后再发送B,但是在异常情况下,在A发送失败的情况下,B发送成功,而A由于重试机制在B发送完成之后重试发送成功了。这时对于本身顺序为AB的消息顺序变成了BA。

针对这种问题,严格的顺序消费还需要max.in.flight.requests.per.connection参数的支持。

该参数指定了生产者在收到服务器响应之前可以发送多少个消息。它的值越高,就会占用越多的内存,同时也会提升吞吐量。把它设为1就可以保证消息是按照发送的顺序写入服务器的。

此外,对于某些业务场景,设置max.in.flight.requests.per.connection=1会严重降低吞吐量,如果放弃使用这种同步重试机制,则可以考虑在消费端增加失败标记的记录,然后用定时任务轮询去重试这些失败的消息并做好监控报警。

0条评论
作者已关闭评论
肖****凯
4文章数
0粉丝数
肖****凯
4 文章 | 0 粉丝
肖****凯
4文章数
0粉丝数
肖****凯
4 文章 | 0 粉丝
原创

kafka如何保证消息的顺序性

2024-08-29 09:42:16
38
0

前提

生产者给kakfa投递的消息,在同一个topic下不同的Partition中,不同的Partition之间的消息是无法保证顺序的。但有些场景,业务需要处理顺序消息。怎么保证kakfa消息的顺序性,以致最终消费者能顺序消费消息呢

消息有序性分类

  • 全局有序:一个Topic下的所有消息都需要按照生产顺序消费。
  • 局部有序:一个Topic下的消息,只需要满足同一业务字段的要按照生产顺序消费。例如:Topic消息是订单的流水表,包含订单orderId,业务要求同一个orderId的消息需要按照生产顺序进行消费。

全局有序

由于Kafka的一个Topic可以分为了多个Partition,Producer发送消息的时候,是分散在不同 Partition的。当Producer按顺序发消息给Broker,但进入Kafka之后,这些消息就不一定进到哪个Partition,会导致顺序是乱的。因此要满足全局有序,需要1个Topic只能对应1个Partition。而且对应的consumer也要使用单线程或者保证消费顺序的线程模型.

局部有序

要满足局部有序,只需要在发消息的时候指定Partition Key,Kafka对其进行Hash计算,根据计算结果决定放入哪个Partition。这样Partition Key相同的消息会放在同一个Partition。确保将相关的消息发送到同一个分区。Kafka 保证在单个分区内消息的顺序。因此,如果消息的顺序性对业务非常重要,可以将相关的消息通过相同的分区键(如用户 ID 或订单 ID)发送至同一分区。此时,Partition的数量仍然可以设置多个,提升Topic的整体吞吐量。

 

提升消费效率

在不增加partition数量的情况下想提高消费速度,可以考虑以上局部有序的场景下再次hash唯一标识(例如订单orderId)到不同的线程上,多个消费者线程并发处理消息(依旧可以保证局部有序)。说到底还是一个实例线程消费一个partition消息

消息重试对顺序消息的影响

对于一个有着先后顺序的消息A、B,正常情况下应该是A先发送完成后再发送B,但是在异常情况下,在A发送失败的情况下,B发送成功,而A由于重试机制在B发送完成之后重试发送成功了。这时对于本身顺序为AB的消息顺序变成了BA。

针对这种问题,严格的顺序消费还需要max.in.flight.requests.per.connection参数的支持。

该参数指定了生产者在收到服务器响应之前可以发送多少个消息。它的值越高,就会占用越多的内存,同时也会提升吞吐量。把它设为1就可以保证消息是按照发送的顺序写入服务器的。

此外,对于某些业务场景,设置max.in.flight.requests.per.connection=1会严重降低吞吐量,如果放弃使用这种同步重试机制,则可以考虑在消费端增加失败标记的记录,然后用定时任务轮询去重试这些失败的消息并做好监控报警。

文章来自个人专栏
视图流媒体
4 文章 | 1 订阅
0条评论
作者已关闭评论
作者已关闭评论
0
0