当前位置：首页 > news >正文

RocketMQ之偏移量Offset

news 来源：原创 2025/5/11 10:43:21

偏移量：偏移量是 ConsumerQueue 文件中的逻辑位置，表示消息在队列中的顺序。每次消费通过偏移量在索引文件ConsumerQueue中快速找到对应的消息内容。

1.Broker端

Broker端主要通过ConsumerManageProcessor类来处理Offset的更新查询的请求，真正实现Offset的更新查询动作的是类ConsumerOffsetManager，其父类ConfigManager负责将Offset相关信息持久化。

持久化：每隔5秒的定时任务【BrokerController#initializeBrokerScheduledTasks】将ConsumerOffsetManager中的offsetTable【jvm内存 ~ Map集合类型】持久化到consumerOffset.json文件中。

初始化：Broker端启动时借助ConfigManager类将consumerOffset.json文件中偏移量信息加载到类ConsumerOffsetManager中的Map类型集合offsetTable中。

public class ConsumerManageProcessor implements NettyRequestProcessor {public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)throws RemotingCommandException {switch (request.getCode()) {case RequestCode.GET_CONSUMER_LIST_BY_GROUP:return this.getConsumerListByGroup(ctx, request);case RequestCode.UPDATE_CONSUMER_OFFSET://更新偏移量return this.updateConsumerOffset(ctx, request);case RequestCode.QUERY_CONSUMER_OFFSET://查询偏移量return this.queryConsumerOffset(ctx, request);default:break;}return null;}private RemotingCommand queryConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {final RemotingCommand response = RemotingCommand.createResponseCommand(QueryConsumerOffsetResponseHeader.class);final QueryConsumerOffsetResponseHeader responseHeader = (QueryConsumerOffsetResponseHeader) response.readCustomHeader();final QueryConsumerOffsetRequestHeader requestHeader =(QueryConsumerOffsetRequestHeader) request.decodeCommandCustomHeader(QueryConsumerOffsetRequestHeader.class);TopicQueueMappingContext mappingContext = this.brokerController.getTopicQueueMappingManager().buildTopicQueueMappingContext(requestHeader);RemotingCommand rewriteResult = rewriteRequestForStaticTopic(requestHeader, mappingContext);if (rewriteResult != null) {return rewriteResult;}// offset = -1 表明首次消费ConsumerOffsetManager consumerOffsetManager = this.brokerController.getConsumerOffsetManager();String consumerGroup = requestHeader.getConsumerGroup();String topic = requestHeader.getTopic();// 从jvm内存中获取long offset = consumerOffsetManager.queryOffset(consumerGroup, topic, requestHeader.getQueueId());MessageStore messageStore = this.brokerController.getMessageStore();if (offset >= 0) {responseHeader.setOffset(offset);response.setCode(ResponseCode.SUCCESS);response.setRemark(null);} else {long minOffset = messageStore.getMinOffsetInQueue(requestHeader.getTopic(),requestHeader.getQueueId());Boolean setZeroIfNotFound = requestHeader.getSetZeroIfNotFound();if (requestHeader.getSetZeroIfNotFound() != null && Boolean.FALSE.equals(setZeroIfNotFound)) {response.setCode(22);response.setRemark("Not found, do not set to zero, maybe this group boot first");} else if (minOffset <= 0 && messageStore.checkInMemByConsumeOffset(requestHeader.getTopic(), requestHeader.getQueueId(), 0, 1)) {responseHeader.setOffset(0L);response.setCode(ResponseCode.SUCCESS);response.setRemark(null);} else {response.setCode(22);response.setRemark("Not found, V3_0_6_SNAPSHOT maybe this group consumer boot first");}}RemotingCommand rewriteResponseResult = rewriteResponseForStaticTopic(requestHeader, responseHeader, mappingContext, response.getCode());if (rewriteResponseResult != null) {return rewriteResponseResult;}return response;}
}

如上伪代码所示：Broker返回的偏移量大于等于0。

2.消费端

2.1.远程同步偏移量

public class MQClientInstance {ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(r -> new Thread(r));private void startScheduledTask() {...this.scheduledExecutorService.scheduleAtFixedRate(() -> {MQClientInstance.this.persistAllConsumerOffset();}, 1000 * 10, 1000 * 5, TimeUnit.MILLISECONDS);//每隔5秒...}
}

如上伪代码所示：每隔5秒将RemoteBrokerOffsetStore类中offsetTable维护的偏移量同步至Broker端，保证消费端 <-> Broker端偏移量的一致性。

2.2.远程拉取消费数据

public class DefaultMQPushConsumerImpl implements MQConsumerInner {public void pullMessage(final PullRequest pullRequest) {final ProcessQueue processQueue = pullRequest.getProcessQueue();if (!this.consumeOrderly) {...} else {if (processQueue.isLocked()) {if (!pullRequest.isLockedFirst()) {//远程获取偏移量final long offset = this.rebalanceImpl.computePullFromWhere(pullRequest.getMessageQueue());boolean brokerBusy = offset < pullRequest.getNextOffset();pullRequest.setLockedFirst(true);pullRequest.setNextOffset(offset);}} else {this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);log.info("pull message later because not locked in broker, {}", pullRequest);return;}}...PullCallback pullCallback = new PullCallback() {@Overridepublic void onSuccess(PullResult pullResult) {if (pullResult != null) {pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,subscriptionData);switch (pullResult.getPullStatus()) {case FOUND://成功拉取到数据long prevRequestOffset = pullRequest.getNextOffset();// 设置下次消费需要的偏移量pullRequest.setNextOffset(pullResult.getNextBeginOffset());long firstMsgOffset = Long.MAX_VALUE;if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);} else {firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();...boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());//pullResult.getMsgFoundList()是指远程拉取的数据。将目标数据抽象为Runnable类型的ConsumeRequestDefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(pullResult.getMsgFoundList(),processQueue,pullRequest.getMessageQueue(),dispatchToConsume);...}break;...default:break;}}}};...//触发消费端拉取数据的请求this.pullAPIWrapper.pullKernelImpl(...);}
}

public class ConsumeMessageOrderlyService implements ConsumeMessageService {class ConsumeRequest implements Runnable {private final ProcessQueue processQueue;private final MessageQueue messageQueue;public ConsumeRequest(ProcessQueue processQueue, MessageQueue messageQueue) {this.processQueue = processQueue;this.messageQueue = messageQueue;}@Overridepublic void run() {for (boolean continueConsume = true; continueConsume; ) {...if (!msgs.isEmpty()) {final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);...long beginTimestamp = System.currentTimeMillis();ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;boolean hasException = false;this.processQueue.getLockConsume().lock();//回调用户自定义的监听器status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);...// 此步骤会更新 RemoteBrokerOffsetStore#offsetTable集合中当前队列对应的偏移量continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);} else {continueConsume = false;}}}}    
}

public class RebalancePushImpl extends RebalanceImpl {@Overridepublic 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: {long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);if (lastOffset >= 0) {//如果broker返回的偏移量大于0，表明已经被消费过result = lastOffset;}// First start,no offsetelse if (-1 == lastOffset) {if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {result = 0L;} else {//如果返回-1则选择从最大偏移量开始消费result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);}} else {result = -1;}break;}case CONSUME_FROM_FIRST_OFFSET: {long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);if (lastOffset >= 0) {//如果broker返回的偏移量大于0，表明已经被消费过result = lastOffset;} else if (-1 == lastOffset) {//如果返回-1则继续选择从偏移量0开始消费result = 0L;} else {result = -1;}break;}case CONSUME_FROM_TIMESTAMP: {long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);if (lastOffset >= 0) {//如果broker返回的偏移量大于0，表明已经被消费过result = lastOffset;} else if (-1 == lastOffset) {if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);} else {// 获取自定义的时间戳，默认时间节点为：半小时之前【距离当前】long timestamp = UtilAll.parseDate(this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeTimestamp(),UtilAll.YYYYMMDDHHMMSS).getTime();result = this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);}} else {result = -1;}break;}default:break;}return result;}
}

消费端远程拉取消息过程中通过如上伪代码获取当前消息的偏移量。

1.Broker端

2.消费端

2.1.远程同步偏移量

2.2.远程拉取消费数据

相关文章：