Pulsar Consumer redeliver 实现逻辑

2019-07-12  本文已影响0人  wolf4j

Consumer redeliver

在 consumer 中,redeliver的逻辑触发主要有两种形式,第一种是通过用户在创建consumer时,指定 acktimeout 的时间,一种是手动调用consumer端暴漏给用户的 redeliver 接口。

Consumer UnackTracker

在当前版本(2.3.2)中,consumer UnackTracker 只在 Sharedkey_Shared 这两种订阅模型中使用。因为经过 UnackTracker 之后,消息的顺序是没办法被保证的。目前,ExclusiveFailover 这两种订阅模型是没有 UnackTracker 的逻辑,当 ackTimeout 到达之后,broker 会找到当前没有被 unack 的那个消息的位置,将消息重新给发送过来。在 Sharedkey_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 的逻辑。

message

redeliverUnacknowledgedMessages 有两个接口,一个是有参数的,一个没有参数,他们之间的主要区别在于:

当有参数的时候,redeliver的messages是从consumer端的unackMessageTracker中传入进来的,当没有参数的时候,redeliver的messages是从broker端的pendingAck中获取到的。

总结:

redeliver的逻辑需要按照不同的订阅类型来讲述,主要分为如下两种:

上一篇下一篇

猜你喜欢

热点阅读