RocketMQ 消息回溯实践与解析

public static void main(String[] args) throws MQClientException, InterruptedException {        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");        producer.setNamesrvAddr("127.0.0.1:9876");        producer.start();        for (int i = 0; i < 10000; i++) {            try {                Message msg = new Message("TopicTest" /* Topic */,                    "TagA" /* Tag */,                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */                );                SendResult sendResult = producer.send(msg);                System.out.printf("%s%n", sendResult);            } catch (Exception e) {                e.printStackTrace();                Thread.sleep(1000);            }        }        producer.shutdown();    }

public static void main(String[] args) throws InterruptedException, MQClientException {    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");    consumer.setNamesrvAddr("127.0.0.1:9876");    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);    consumer.subscribe("TopicTest", "*");        final AtomicInteger count = new AtomicInteger();        consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,            ConsumeConcurrentlyContext context) {            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);            count.incrementAndGet();            System.out.printf("%s Receive New Messages End: %s %n", Thread.currentThread().getName(), msgs);            System.out.println(count.get());            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;        }    });
    consumer.start();}

mqadmin.cmd resetOffsetByTime -n 127.0.0.1:9876 -t TopicTest -g please_rename_unique_group_name_4 -s 1722240069000

public static void main(String[] args) {    String[] params = new String[]{"resetOffsetByTime","-n","127.0.0.1:9876","-t", "TopicTest", "-g", "please_rename_unique_group_name_4", "-s", "1722240069000"};    MQAdminStartup.main(params);}

/*根据动作标识解析除对应的处理类，我们本次请求实际处理策略类：ResetOffsetByTimeCommand*/SubCommand cmd = findSubCommand(args[0]);/*解析命令行*/Options options = ServerUtil.buildCommandlineOptions(new Options());CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin " + cmd.commandName(), subargs, cmd.buildCommandlineOptions(options),                            new PosixParser());                            /*提交请求执行*/cmd.execute(commandLine, options, rpcHook);

public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException {    DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);        String group = commandLine.getOptionValue("g").trim();//消费者组    String topic = commandLine.getOptionValue("t").trim();//主题    String timeStampStr = commandLine.getOptionValue("s").trim();//重置时间戳    long timestamp = timeStampStr.equals("now") ? System.currentTimeMillis() : Long.parseLong(timeStampStr);//重置时间戳    boolean isC = false;//是否C客户端    boolean force = true;//是否强制重置，这里提前解释一下，有可能时间戳对应的offset比当前消费进度要大，强制的话会出现部分消息消费不到    if (commandLine.hasOption('f')) {        force = Boolean.valueOf(commandLine.getOptionValue("f").trim());    }
    /*与nameserver以及broker交互的客户端启动*/    defaultMQAdminExt.start();    /*正式执行命令*/    Map<MessageQueue, Long> offsetTable = defaultMQAdminExt.resetOffsetByTimestamp(topic, group, timestamp, force, isC);}

public Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce,    boolean isC)    throws RemotingException, MQBrokerException, InterruptedException, MQClientException {    /*从nameserver处获取broker地址*/    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);    /*由于消息数据分区分片，topic下的messagequeue可能存在多个broker上，因此这是个列表*/    List<BrokerData> brokerDatas = topicRouteData.getBrokerDatas();    Map<MessageQueue, Long> allOffsetTable = new HashMap<MessageQueue, Long>();    if (brokerDatas != null) {        for (BrokerData brokerData : brokerDatas) {            String addr = brokerData.selectBrokerAddr();            if (addr != null) {                /*循环与各个broker交互，执行重置操作*/                Map<MessageQueue, Long> offsetTable =                    this.mqClientInstance.getMQClientAPIImpl().invokeBrokerToResetOffset(addr, topic, group, timestamp, isForce,                        timeoutMillis, isC);                if (offsetTable != null) {                    allOffsetTable.putAll(offsetTable);                }            }        }    }    return allOffsetTable;}

public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis,    boolean allowTopicNotExist) throws MQClientException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {    GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader();    requestHeader.setTopic(topic); /*同样的组装参数，请求码：105*/    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINTO_BY_TOPIC, requestHeader);
    /*创建请求与nameserver交互*/    RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);    byte[] body = response.getBody();    if (body != null) {        return TopicRouteData.decode(body, TopicRouteData.class);    }}

public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,    RemotingCommand request) throws RemotingCommandException {    final RemotingCommand response = RemotingCommand.createResponseCommand(null);    final GetRouteInfoRequestHeader requestHeader =        (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
    /*nameserver内部存储topic的路由信息*/    TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic()); byte[] content = topicRouteData.encode();    response.setBody(content);    response.setCode(ResponseCode.SUCCESS);    response.setRemark(null);    return response;}

//topic路由信息，根据这个做负载均衡，QueueData里面记录brokerNameprivate final HashMap<String/* topic */, List<QueueData>> topicQueueTable;//broke基本信息 名称  所在集群信息   主备broke地址  brokerId=0表示master   >0表示slaveprivate final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;//集群信息，包含集群所有的broke信息private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;//存活的broke信息，以及对应的channelprivate final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;//broke的过滤类信息private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

public Map<MessageQueue, Long> invokeBrokerToResetOffset(final String addr, final String topic, final String group,    final long timestamp, final boolean isForce, final long timeoutMillis, boolean isC)    throws RemotingException, MQClientException, InterruptedException {        ResetOffsetRequestHeader requestHeader = new ResetOffsetRequestHeader();    requestHeader.setTopic(topic);    requestHeader.setGroup(group);    requestHeader.setTimestamp(timestamp);    requestHeader.setForce(isForce);
    /*同样的组装参数，请求码：222*/    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.INVOKE_BROKER_TO_RESET_OFFSET, requestHeader);    if (isC) {        request.setLanguage(LanguageCode.CPP);    } /*创建请求与broker交互,注意这里是同步invokeSync*/    RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr), request, timeoutMillis);    if (response.getBody() != null) {        ResetOffsetBody body = ResetOffsetBody.decode(response.getBody(), ResetOffsetBody.class);        return body.getOffsetTable();    }}

public RemotingCommand resetOffset(String topic, String group, long timeStamp, boolean isForce,    boolean isC) {    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(topic);
    /*记录下该消费者组消费topic下的队列要重置到哪条offset*/    Map<MessageQueue/*队列*/, Long/*offser*/> offsetTable = new HashMap<MessageQueue, Long>();
    for (int i = 0; i < topicConfig.getWriteQueueNums(); i++) {        MessageQueue mq = new MessageQueue();        mq.setBrokerName(this.brokerController.getBrokerConfig().getBrokerName());        mq.setTopic(topic);        mq.setQueueId(i);
        /*broker可以获取该topic下的consumergroup下的某个队列的offset*/        long consumerOffset =            this.brokerController.getConsumerOffsetManager().queryOffset(group, topic, i);//消费者组当前已经消费的offset        if (-1 == consumerOffset) {            response.setCode(ResponseCode.SYSTEM_ERROR);            response.setRemark(String.format("THe consumer group <%s> not exist", group));            return response;        }
        long timeStampOffset;        if (timeStamp == -1) {   //没有指定表示当前队列最大的offset            timeStampOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);        } else {            //根据时间戳查到队列下对应的offset            timeStampOffset = this.brokerController.getMessageStore().getOffsetInQueueByTime(topic, i, timeStamp);        }
        if (timeStampOffset < 0) {            //<0表示消息已经被删掉了            log.warn("reset offset is invalid. topic={}, queueId={}, timeStampOffset={}", topic, i, timeStampOffset);            timeStampOffset = 0;        }
        /*如果isForce=false，则要重置的offset<当前正在消费的offset才会重置。也过来，也就是说重置不仅会回溯，消费进度过慢也可以往后拨，加快消费进度*/        if (isForce || timeStampOffset < consumerOffset) {            offsetTable.put(mq, timeStampOffset);        } else {            offsetTable.put(mq, consumerOffset);        }    }
    /*确定了要先重置的offset之后开始与客户端交互，准备客户端重置，请求码220*/    ResetOffsetRequestHeader requestHeader = new ResetOffsetRequestHeader();    requestHeader.setTopic(topic);    requestHeader.setGroup(group);    requestHeader.setTimestamp(timeStamp);    RemotingCommand request =        RemotingCommand.createRequestCommand(RequestCode.RESET_CONSUMER_CLIENT_OFFSET, requestHeader);    if (isC) {        // c++ language        ResetOffsetBodyForC body = new ResetOffsetBodyForC();        List<MessageQueueForC> offsetList = convertOffsetTable2OffsetList(offsetTable);        body.setOffsetTable(offsetList);        request.setBody(body.encode());    } else {        // other language        ResetOffsetBody body = new ResetOffsetBody();        body.setOffsetTable(offsetTable);        request.setBody(body.encode());    }
    /*拿到与当前broker建立连接的消费者组客户端信息*/    ConsumerGroupInfo consumerGroupInfo =        this.brokerController.getConsumerManager().getConsumerGroupInfo(group);
    if (consumerGroupInfo != null && !consumerGroupInfo.getAllChannel().isEmpty()) {        //获取长连接channel        ConcurrentMap<Channel, ClientChannelInfo> channelInfoTable =            consumerGroupInfo.getChannelInfoTable();        for (Map.Entry<Channel, ClientChannelInfo> entry : channelInfoTable.entrySet()) {            int version = entry.getValue().getVersion();            /*这里版本可以判断，只有客户端版本>3.0.7才支持重置*/            if (version >= MQVersion.Version.V3_0_7_SNAPSHOT.ordinal()) {                try {                    /*注意这里是只管发不管收，可以简单理解为异步了invokeOneway*/                    this.brokerController.getRemotingServer().invokeOneway(entry.getKey(), request, 5000);                    log.info("[reset-offset] reset offset success. topic={}, group={}, clientId={}",                        topic, group, entry.getValue().getClientId());                } catch (Exception e) {                    log.error("[reset-offset] reset offset exception. topic={}, group={}",                        new Object[] {topic, group}, e);                }            } else {                response.setCode(ResponseCode.SYSTEM_ERROR);                response.setRemark("the client does not support this feature. version="                    + MQVersion.getVersionDesc(version));                log.warn("[reset-offset] the client does not support this feature. version={}",                    RemotingHelper.parseChannelRemoteAddr(entry.getKey()), MQVersion.getVersionDesc(version));                return response;            }        }    } else {        String errorInfo =            String.format("Consumer not online, so can not reset offset, Group: %s Topic: %s Timestamp: %d",                requestHeader.getGroup(),                requestHeader.getTopic(),                requestHeader.getTimestamp());        log.error(errorInfo);        response.setCode(ResponseCode.CONSUMER_NOT_ONLINE);        response.setRemark(errorInfo);        return response;    }    response.setCode(ResponseCode.SUCCESS);    ResetOffsetBody resBody = new ResetOffsetBody();    resBody.setOffsetTable(offsetTable);    response.setBody(resBody.encode());    return response;}

public void resetOffset(String topic, String group, Map<MessageQueue, Long> offsetTable) {    DefaultMQPushConsumerImpl consumer = null;    try {        /*根据消费者组找到对应的消费实现，即我们熟悉的DefaultMQPushConsumerImpl或者DefaultMQPullConsumerImpl*/        MQConsumerInner impl = this.consumerTable.get(group);        if (impl != null && impl instanceof DefaultMQPushConsumerImpl) {            consumer = (DefaultMQPushConsumerImpl) impl;        } else {            //由于PullConsumer消费进度自己控制，因此直接返回            log.info("[reset-offset] consumer dose not exist. group={}", group);            return;        }                consumer.suspend();//暂停消费
        /*暂停消息拉取，以及待处理的消息缓存都清掉*/        ConcurrentMap<MessageQueue, ProcessQueue> processQueueTable = consumer.getRebalanceImpl().getProcessQueueTable();        for (Map.Entry<MessageQueue, ProcessQueue> entry : processQueueTable.entrySet()) {            MessageQueue mq = entry.getKey();            if (topic.equals(mq.getTopic()) && offsetTable.containsKey(mq)) {                ProcessQueue pq = entry.getValue();                pq.setDropped(true);                pq.clear();            }        }          /*这里的等待实现比较简单，与broker交互是同步，broker与consumer交互是异步，因此这里阻塞10秒是为了保证所有的consumer都在这里存储offset并触发reblance*/        try {            TimeUnit.SECONDS.sleep(10);        } catch (InterruptedException e) {        }
        Iterator<MessageQueue> iterator = processQueueTable.keySet().iterator();        while (iterator.hasNext()) {            MessageQueue mq = iterator.next();            //获取messagequeue应该被重置的offset            Long offset = offsetTable.get(mq);            if (topic.equals(mq.getTopic()) && offset != null) {                try {                    /*更新更新本地offset，这里注意集群模式是先修改本地，然后定时任务每五秒上报broker，而广播模式offset在本地存储，因此只需要修改消费者本地的offset即可*/                    consumer.updateConsumeOffset(mq, offset);                    consumer.getRebalanceImpl().removeUnnecessaryMessageQueue(mq, processQueueTable.get(mq));                    iterator.remove();                } catch (Exception e) {                    log.warn("reset offset failed. group={}, {}", group, mq, e);                }            }        }    } finally {        if (consumer != null) {            /*重新触发reblance，由于broker已经重置的该消费者组的offset，因此重分配后以broker为准*/            consumer.resume();        }    }}

public class PullConsumerLocalTest {    private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<MessageQueue, Long>();    private static final Map<MessageQueue, Pair<Long/*最小offset*/,Long/*最大offset*/>> QUEUE_OFFSE_SECTION_TABLE = new HashMap<>();    private static final Long MIN_TIMESTAMP = 1722240069000L;//最小时间戳    private static final Long MAX_TIMESTAMP = 1722240160000L;//最大时间戳
    public static void main(String[] args) throws MQClientException {        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");        consumer.setNamesrvAddr("127.0.0.1:9876");        consumer.start();
        /*初始化待处理的offset*/        String topic = "TopicTest";        init(consumer, topic);
        Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(topic);        for (MessageQueue mq : mqs) {            System.out.printf("Consume from the queue: %s%n", mq);            SINGLE_MQ:            while (true) {                try {                    PullResult pullResult =                        consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);                    System.out.printf("%s%n", pullResult);                    putMessageQueueOffset(mq, pullResult.getNextBeginOffset());                    switch (pullResult.getPullStatus()) {                        case FOUND:                            //check max offset and dosomething...                            break;                        case NO_MATCHED_MSG:                            break;                        case NO_NEW_MSG:                            break SINGLE_MQ;                        case OFFSET_ILLEGAL:                            break;                        default:                            break;                    }                } catch (Exception e) {                    e.printStackTrace();                }            }        }
        consumer.shutdown();    }
    private static void init(DefaultMQPullConsumer consumer, String topic) throws MQClientException {        Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(topic);        for (MessageQueue mq : mqs) {            long minOffset = consumer.searchOffset(mq, MIN_TIMESTAMP);            long maxOffset = consumer.searchOffset(mq, MAX_TIMESTAMP);            //记录区间内范围内最小以及最大的offset            QUEUE_OFFSE_SECTION_TABLE.put(mq, new Pair<>(minOffset, maxOffset));            //将最小offset写为下次消费的初始offset            OFFSE_TABLE.put(mq, minOffset);        }    }
    private static long getMessageQueueOffset(MessageQueue mq) {        Long offset = OFFSE_TABLE.get(mq);        if (offset != null)            return offset;
        return 0;    }
    private static void putMessageQueueOffset(MessageQueue mq, long offset) {        OFFSE_TABLE.put(mq, offset);    }
}