Consumer redeliver
在 consumer 中,redeliver的逻辑触发主要有两种形式,第一种是通过用户在创建consumer时,指定 acktimeout
的时间,一种是手动调用consumer端暴漏给用户的 redeliver 接口。
Consumer UnackTracker
在当前版本(2.3.2)中,consumer UnackTracker 只在 Shared
和 key_Shared
这两种订阅模型中使用。因为经过 UnackTracker 之后,消息的顺序是没办法被保证的。目前,Exclusive
和 Failover
这两种订阅模型是没有 UnackTracker 的逻辑,当 ackTimeout 到达之后,broker 会找到当前没有被 unack 的那个消息的位置,将消息重新给发送过来。在 Shared
和 key_Shared
这两种订阅模型下,对于没有 ack 的消息,会将他们 tracker 到一个集合(例如:set)中,当达到了用户设置的 ack timeout 之后,被tracker 的消息如果还没有被 ack ,会将整个 tracker 的消息,重新 redeliver 给当前的 consumer。
在触发 redeliver 的逻辑时,需要使用到 CommandRedeliverUnacknowledgedMessages
,告诉 broker 可以 redeliver 消息过来,proto文件定义如下:
message CommandRedeliverUnacknowledgedMessages {
required uint64 consumer_id = 1;
repeated MessageIdData message_ids = 2;
}
CommandRedeliverUnacknowledgedMessages
包含两个字段,一个是 consumer_id
表示需要将消息重新发送给哪个 consumer,一个是 message_ids
,表示需要发送哪些消息到当前的 consumer 中。需要进一步说明的是,MessageIdData
有四个字段构成,在这里比较主要的是在多 partition 的情况下,需要根据其中的 partition_id 进一步来确定,哪些 messages 是属于当前 partition 的 consumer。
源码实现
最直接的,用来与 client 端交互的是在 serverCnx
下的 handleRedeliverUnacknowledged
这个函数,如下:
protected void handleRedeliverUnacknowledged(CommandRedeliverUnacknowledgedMessages redeliver) {
// 判断当前的连接状态
checkArgument(state == State.Connected);
if (log.isDebugEnabled()) {
log.debug("[{}] Received Resend Command from consumer {} ", remoteAddress, redeliver.getConsumerId());
}
// 获取consumer_id
CompletableFuture<Consumer> consumerFuture = consumers.get(redeliver.getConsumerId());
if (consumerFuture != null && consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {
Consumer consumer = consumerFuture.getNow(null);
// Subscription.isIndividualAckMode() 这个函数用来判断当前 consumer 订阅的类型是否是 Shared 或者 KeyShared。
if (redeliver.getMessageIdsCount() > 0 && Subscription.isIndividualAckMode(consumer.subType())) {
consumer.redeliverUnacknowledgedMessages(redeliver.getMessageIdsList());
} else {
// 如果订阅类型为 Exclusive 或者 Failover,执行下面 redeliver 的逻辑
consumer.redeliverUnacknowledgedMessages();
}
}
}
下面我们来看一下 redeliverUnacknowledgedMessages 的逻辑。
redeliverUnacknowledgedMessages 有两个接口,一个是有参数的,一个没有参数,他们之间的主要区别在于:
当有参数的时候,redeliver的messages是从consumer端的unackMessageTracker中传入进来的,当没有参数的时候,redeliver的messages是从broker端的pendingAck中获取到的。
总结:
redeliver的逻辑需要按照不同的订阅类型来讲述,主要分为如下两种:
- Exclusive and Filover
- 这种逻辑相对比较简单,因为他是需要保证顺序的,所以每次redeliver就从当前没有被unack的那个位置点开始将消息发送给client端,具体消息是否被ack取决于client端是否将该条消息的ACK 命令发送给broker端
- Shared and key_shared
- 这种逻辑相对比较复杂,因为他是无序的状态,所以我需要去tracker每一条消息。
- 所以在broker端有一个pendingAck的集合,当消息从producer端发送到broker中的ledger中之后,会有一个线程异步的去将ledger中的消息读取到pendingAck的集合中,这个时候ledger中的cursor位置会跟着向前移动,这个是broker端的逻辑。
- 在client端,在consumer中会有一个unackmessageTracker的集合,至于这个集合什么时间开始tracker每一条消息,需要明白一个逻辑,一条消息被成功接收总共分为两步,第一步:receive;第二步:ack。unackmessageTracker所关心的消息是那些被consumer成功receiver到但是没有被ack的消息。当acktimeout到期之后,如果消息还没有被ack的就会去redeliver这部分被tracker的数据。在一个多partition的topic中,有好几个consumer,每一个consumer都有自己对应的一个unackmessageTracker的set集合,所以,他们就类似与pendingAck的子集。这些tracker的消息正常情况下一定会在pendingAck的集合中被找到。
- unacktracker中有一个time的触发器,触发的时间就是acktimeout设定的时间。
- redeliver messages 有两种形式,第一种,你可以将自己本地的tracker的消息发送给broker,broker会去给你redeliver相应的消息。第二种,发送一个空的redeliver命令上去,此时,broker会将pendingACK中所有的消息全部给你redeliver到consumer。