rocketmq源碼解析消費者管理處理器②

時間 2019-11-11

標籤 rocketmq 源碼解析消費者管理處理器简体版

原文原文鏈接

說在前面java

消費者管理處理器，更新消費者的offsetapache

源碼解析微信

進入這個方法，更新消費者的offset，org.apache.rocketmq.broker.processor.ConsumerManageProcessor#updateConsumerOffsetless

private RemotingCommand updateConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request)throws RemotingCommandException {final RemotingCommand response =RemotingCommand.createResponseCommand(UpdateConsumerOffsetResponseHeader.class);final UpdateConsumerOffsetRequestHeader requestHeader =(UpdateConsumerOffsetRequestHeader) request.decodeCommandCustomHeader(UpdateConsumerOffsetRequestHeader.class);//        提交offset=》this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), requestHeader.getConsumerGroup(),requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset());response.setCode(ResponseCode.SUCCESS);response.setRemark(null);return response;    }

進入這個方法，提交offset，org.apache.rocketmq.broker.offset.ConsumerOffsetManager#commitOffset(java.lang.String, java.lang.String, java.lang.String, int, long)this

public void commitOffset(final String clientHost, final String group, final String topic, final int queueId,final long offset) {// topic@groupString key = topic + TOPIC_GROUP_SEPARATOR + group;//        =》this.commitOffset(clientHost, key, queueId, offset);    }

進入這個方法，org.apache.rocketmq.broker.offset.ConsumerOffsetManager#commitOffset(java.lang.String, java.lang.String, int, long)code

private void commitOffset(final String clientHost, final String key, final int queueId, final long offset) {ConcurrentMap<Integer, Long> map = this.offsetTable.get(key);if (null == map) {map = new ConcurrentHashMap<Integer, Long>(32);map.put(queueId, offset);this.offsetTable.put(key, map);} else {Long storeOffset = map.put(queueId, offset);if (storeOffset != null && offset < storeOffset) {log.warn("[NOTIFYME]update consumer offset less than store. clientHost={}, key={}, queueId={}, requestOffset={}, storeOffset={}", clientHost, key, queueId, offset, storeOffset);}}    }

返回到這個方法，org.apache.rocketmq.broker.processor.ConsumerManageProcessor#updateConsumerOffset結束。blog

說在最後get

本次解析僅表明我的觀點，僅供參考。源碼

加入技術微信羣it