文档章节

Apache pulsar 技术系列-- 消息重推的几种方式

腾讯云中间件
 腾讯云中间件
发布于 2023/07/26 17:32
字数 1578
阅读 1.8K
收藏 0

导语

Apache Pulsar 是一个多租户、高性能的服务间消息传输解决方案,支持多租户、低延时、读写分离、跨地域复制(GEO replication)、快速扩容、灵活容错等特性。在很多场景下,用户需要通过 MQ 实现消息的重新推送能力,比如超时重推、处理异常时重推等,本文介绍 Apache Pulsar 提供的几种消息重推方案。

在 MQ 实际的使用中,用户消费数据时,可能会遇到消息处理异常或者需要推迟处理的场景,这里就涉及到消息的重推逻辑,Pulsar 自己提供了消息重推的能力。本文主要介绍 Pulsar 的消息重推机制。

消息获取(拉取/推送)机制

Pulsar 的消费采用了推、拉结合的消息获取机制,Consumer 获取消息之前会首先通知 Broker(FLOW 请求),Broker 会根据配置的 ReceiveQueue 大小以及 Consumer 当前可以接收的消息数量来推送消息给 Consumer。

详细的交互流程如下图所示:

  1. Consumer 在创建之后,会以 MaxReceiveQueue 的大小作为 Permit 值,这个值就是 Consumer 可以缓存的的最大消息条数。

  2. 然后,Consumer 向 Broker 发起 FLOW 请求,携带 Permit 信息(Consumer Permit 减少到 0),Broker 接收之后会记录这个 Permit 作为 Consumer 的 AvailablePermit,AvailablePermit 决定 Broker 可以向 Consumer 发送数据的数量(实际是在读取数据时判断)。

  3. 如果 AvailablePermit > 0, Broker 开始读取数据(假设有 N 条),然后推送给 Consumer,推送之后,AvailablePermit 自减 N。

  4. Consumer 接收到消息之后,并不会直接返回给用户,而是放在 ReceiveQueue 中,当用户调用 Receive() 方法来获取消息时,Consumer 将 Permit + 1。

  5. 当 Permit > MaxReceiveQueueSize / 2,Consumer 会再次发起 Flow 请求,并且携带当前的 Permit 值。

上述流程,就是 Consumer 和 Broker 的消息传递过程。

在默认的情况下,数据推送给 Consumer 之后,就完全交给用户处理,数据不会重复推送。这种方式满足不了需要重推的场景,下面介绍目前 Pulsar 的几种重推机制。

SDK 统一的重推

一个比较直观的做法是超过一定时间,如果消息没有 Ack 就重新推送。

目前 Pulsar 提供了通过超时时间来控制数据重推的能力,Consumer 可以配置 AckTimeout(默认关闭),在设置了 AckTimeout 之后,Client 会构建一个 UnAckedMessageTracker ,用户 Receive() 的所有的消息都会被 UnAckedMessageTracker 跟踪。用户 Ack 消息时,会从 UnAckedMessageTracker 删除,对于没有 Ack 的消息,UnAckedMessageTracker 会有定时任务来检查,如果已经超过了 AckTimeout 时间,则会触发重推。

重推是通过 RedeliverUnackMessage 来实现的,UnAckedMessageTracker 会主动发起 Redeliver 的请求,Broker 会根据请求的 MessageId 信息重新推送。

AckTimeout 在 Consumer 初始化时设置:

Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                                                              .ackTimeout(10, TimeUnit.SECOND)

用户决策的重推 -- NegativeAck

通过 AckTimeout 实现的重推,是 SDK 内部统一实现的,用户不能控制重推的行为,如果用户希望根据自己的使用场景,决定哪些消息需要重推,Pulsar 提供了 NegativeAck 的能力。

NegativeAck 和 AckTimeout 方式类似,有一个 NegativeAcksTracker 来管理消息的重推,NegativeAcksTracker 只会跟踪用户主动调用 NegativeAcknowledge() 方法的 MessageID,重推的逻辑也是通过 RedeliverUnackMessage 实现。

NegativeAck 可以设置 Redelivery 的 Delay 时间。

 Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                              .negativeAckRedeliveryDelay(1001, TimeUnit.MILLISECONDS)

使用的时候,需要明确调用。

// call the API to send negative 
acknowledgmentconsumer.negativeAcknowledge(message);

用户决策的重推 -- RLQ

除了 NegativeAck 的方式,用户还可以通过重试队列( RLQ )来实现主动的消息重推,RLQ 一般会使用在用户暂时不能处理某些消息,并且希望之后再处理的场景。

Pulsar 提供了 ReconsumeLater() 方法来实现重试队列,和 Negative 不同的是,RLQ 会创建一个新的 Topic,Topic 的格式是 TopicName-SubscriptionName_RLQ , 每次 ReconsumeLater() 时,都会产生一个新的消息写入到 RLQ Topic 中,并且会对之前的消息 Ack。

设置了 RLQ 的 Consumer,SDK 内部默认会启动 RLQ 的订阅,所以 RLQ 的消息也会被 Consumer 消费到。

RLQ 是通过 DeadLetterPolicy 来配置的(DLQ 下文会解释)。

Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
 	  .topic("my-topic")
 	  .subscriptionName("my-subscription")
	  .subscriptionType(SubscriptionType.Shared)
 	  .enableRetry(true)
	  .deadLetterPolicy(DeadLetterPolicy.builder()
 	  .maxRedeliverCount(maxRedeliveryCount)
	  .build())
     .subscribe();

RLQ Topic 中的消息属性中会添加一下信息:

Special property Description
REAL_TOPIC 原始 Topic 名称
ORIGIN_MESSAGE_ID 原始 MessageId
RECONSUMETIMES 重复消费的次数
DELAY_TIME 投递的延迟时间

RLQ 也需要主动调用: consumer.reconsumeLater(msg, 3, TimeUnit.SECONDS)。

为重推次数加上限制--DLQ

对于数据持续处理失败,一直重试并不是一个很好的策略,此时死信队列(DLQ)就是一个比较好的选择,DLQ 允许用户将持续处理失败的数据写入到一个独立的 Dead Letter Topic 中,DLQ 的数据需要单独的订阅来消费。

DLQ Topic 的格式为 TopicName-SubscriptionName_DLQ。DLQ 需要为重试设置一个上限,当重试次数超过上限之后,就会被写入到 DLQ Topic 中。

Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
 		  .topic("my-topic")
		  .subscriptionName("my-subscription")
		  .subscriptionType(SubscriptionType.Shared)
         .deadLetterPolicy(DeadLetterPolicy.builder()
         		.maxRedeliverCount(maxRedeliveryCount)
         		.build())
         .subscribe();

几种重推和 DLQ 的关系

如果配置了 DLQ,那么使用 AckTimeout、NegativeAck 或者 ReconsumeLater 引起的数据重推都会触发 DLQ,也就是说重试的次数达到上限之后,都会被写入到 DLQ topic 里。

重试次数的统计有所区别:

AckTimeout 和 NegativeAck 都是通过 Redelivery 机制来计数的,SDK 发起 Redelivery 请求之后,Broker 侧的 RedeliveryTracker 会记录重推的次数,并且在推送给 Consumer 的 Message 中会包含 RedeliveryCount 的字段。

对于 RLQ,则是从 RECONSUMETIMES 属性中获取重复消费的次数,这个属性在 Client 生成,并且也是在 Client 计数。

总的来说,Apache Pulsar 提供了多种消息重推的方式,用户可以结合自己的场景,灵活使用,满足自己的业务需求。

腾讯云中间件

腾讯云中间件

粉丝 310
博文 300
码字总数 531422
作品 2
深圳
私信 提问
加载中
点击引领话题?
双目视觉理论篇

相机模型与四种参考坐标系 上图中右下角的黑点是真实世界的一个点,最左边的灰色部分是一张数字照片,称为像平面,单位为毫米(mm)。青色的格子则是像平面中一个一个的像素。我们现在需要知道...

算法之名
昨天
51
0
TiDB v7.1.1三地五中心,TiDB POC最佳实践探索

原文来源:https://tidb.net/blog/b4732d88 POC测试背景 在某地震多发省,为了避免地震造成的机房级灾难,或者城市级灾难,导致整个系统不可用,拟建设一套三地五中心五副本分布式高可用数据...

TiDB社区干货传送门
2023/09/28
2
0
首页推荐:彩票导师一对一带上岸带赚钱计划

彩票导师一对一带上岸带赚钱计划?【網子:fc 665? cc】?【 Q?:85,62-329】一对一单带?稳定计划群?最重要还是要找对一个能引领你的人我是这么认为的:心态只能让我们锦上添花,但是并...

osc_4962273979
前天
0
0
Redis 被分叉了。

嘿朋友们。我需要更好地更新这个通讯,而不是在现在每个只有三个用户的五百万个社交平台上随意发布链接。自从我上次发电邮以来,我写了如下内容: 一篇关于GGUF的长篇文章 Check out this g...

vickibo53314cfa41
04/18
11
0
在Go中执行比特币交易的方法

比特币近年来风靡一时。同样,谷歌开发的编程语言Golang也备受关注。今天我将指导您如何使用Golang进行简单的比特币交易。希望在本教程之后,您能够不使用任何DApps进行比特币转账。教程要求...

coderoa69a90f74b5
前天
24
0

没有更多内容

加载失败,请刷新页面

加载更多

{{formatHtml(o.title)}}

{{i}}-{{formatHtml(o.content)}}

{{o.author.name}}
{{o.pubDate | formatDate}}
{{o.viewCount | bigNumberTransform}}
{{o.replyCount | bigNumberTransform}}

暂无文章

便宜云服务器
登录后可查看更多优质内容
返回顶部
顶部
http://www.vxiaotou.com