聊聊rocketmq的ConsumeFromWhere

序

本文主要研究一下rocketmq的ConsumeFromWhere

ConsumeFromWhere

rocketmq-common-4.5.2-sources.jar!/org/apache/rocketmq/common/consumer/ConsumeFromWhere.java

public enum ConsumeFromWhere {
    CONSUME_FROM_LAST_OFFSET,

    @Deprecated
    CONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST,
    @Deprecated
    CONSUME_FROM_MIN_OFFSET,
    @Deprecated
    CONSUME_FROM_MAX_OFFSET,
    CONSUME_FROM_FIRST_OFFSET,
    CONSUME_FROM_TIMESTAMP,
}

• ConsumeFromWhere定義了CONSUME_FROM_LAST_OFFSET、CONSUME_FROM_FIRST_OFFSET、CONSUME_FROM_TIMESTAMP枚舉值

computePullFromWhere

rocketmq-client-4.5.2-sources.jar!/org/apache/rocketmq/client/impl/consumer/RebalancePushImpl.java

public class RebalancePushImpl extends RebalanceImpl {

    //......

    @Override
    public long computePullFromWhere(MessageQueue mq) {
        long result = -1;
        final ConsumeFromWhere consumeFromWhere = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeFromWhere();
        final OffsetStore offsetStore = this.defaultMQPushConsumerImpl.getOffsetStore();
        switch (consumeFromWhere) {
            case CONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST:
            case CONSUME_FROM_MIN_OFFSET:
            case CONSUME_FROM_MAX_OFFSET:
            case CONSUME_FROM_LAST_OFFSET: {
                long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
                if (lastOffset >= 0) {
                    result = lastOffset;
                }
                // First start,no offset
                else if (-1 == lastOffset) {
                    if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                        result = 0L;
                    } else {
                        try {
                            result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
                        } catch (MQClientException e) {
                            result = -1;
                        }
                    }
                } else {
                    result = -1;
                }
                break;
            }
            case CONSUME_FROM_FIRST_OFFSET: {
                long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
                if (lastOffset >= 0) {
                    result = lastOffset;
                } else if (-1 == lastOffset) {
                    result = 0L;
                } else {
                    result = -1;
                }
                break;
            }
            case CONSUME_FROM_TIMESTAMP: {
                long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
                if (lastOffset >= 0) {
                    result = lastOffset;
                } else if (-1 == lastOffset) {
                    if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                        try {
                            result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
                        } catch (MQClientException e) {
                            result = -1;
                        }
                    } else {
                        try {
                            long timestamp = UtilAll.parseDate(this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeTimestamp(),
                                UtilAll.YYYYMMDDHHMMSS).getTime();
                            result = this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);
                        } catch (MQClientException e) {
                            result = -1;
                        }
                    }
                } else {
                    result = -1;
                }
                break;
            }

            default:
                break;
        }

        return result;
    }

    //......
}

• RebalancePushImpl的computePullFromWhere會判斷defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeFromWhere()
• 對於CONSUME_FROM_LAST_OFFSET，若offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE)大於等於0，則更新result爲該值，若lastOffset爲-1，則在mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)時更新result爲0，不然更新result爲mQClientFactory.getMQAdminImpl().maxOffset(mq)
• 對於CONSUME_FROM_FIRST_OFFSET，若offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE)大於等於0，則更新result爲該值，若lastOffset爲-1，則更新result爲0；對於CONSUME_FROM_TIMESTAMP，若offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE)大於等於0，則更新result爲該值；若lastOffset爲-1，則對於mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)取mQClientFactory.getMQAdminImpl().maxOffset(mq)，不然取defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeTimestamp()去搜索QClientFactory.getMQAdminImpl().searchOffset，將返回值更新到result

小結

• ConsumeFromWhere定義了CONSUME_FROM_LAST_OFFSET、CONSUME_FROM_FIRST_OFFSET、CONSUME_FROM_TIMESTAMP枚舉值；RebalancePushImpl的computePullFromWhere會判斷defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeFromWhere()
• 若offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE)大於等於0，則更新result爲該值；對於lastOffset爲-1且mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)，CONSUME_FROM_LAST_OFFSET取0，CONSUME_FROM_TIMESTAMP取mQClientFactory.getMQAdminImpl().maxOffset(mq)
• 對於lastOffset爲-1可是非q.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)的狀況，CONSUME_FROM_LAST_OFFSET取mQClientFactory.getMQAdminImpl().maxOffset(mq)，CONSUME_FROM_TIMESTAMP取mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp)

doc

• ConsumeFromWhere