[RIP-28] light message queue(LMQ) (#3694)

2026-10-29 17:25:35 +08:00 · 2022-01-13 19:15:42 +08:00
parent 8d28d3fa20
commit 14b63ee550
34 changed files with 2466 additions and 283 deletions
@@ -47,11 +47,13 @@ import org.apache.rocketmq.broker.filter.ConsumerFilterManager;
 import org.apache.rocketmq.broker.filtersrv.FilterServerManager;
 import org.apache.rocketmq.broker.latency.BrokerFastFailure;
 import org.apache.rocketmq.broker.latency.BrokerFixedThreadPoolExecutor;
+import org.apache.rocketmq.broker.longpolling.LmqPullRequestHoldService;
 import org.apache.rocketmq.broker.longpolling.NotifyMessageArrivingListener;
 import org.apache.rocketmq.broker.longpolling.PullRequestHoldService;
 import org.apache.rocketmq.broker.mqtrace.ConsumeMessageHook;
 import org.apache.rocketmq.broker.mqtrace.SendMessageHook;
 import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
+import org.apache.rocketmq.broker.offset.LmqConsumerOffsetManager;
 import org.apache.rocketmq.broker.out.BrokerOuterAPI;
 import org.apache.rocketmq.broker.plugin.MessageStoreFactory;
 import org.apache.rocketmq.broker.plugin.MessageStorePluginContext;
@@ -64,7 +66,9 @@ import org.apache.rocketmq.broker.processor.QueryMessageProcessor;
 import org.apache.rocketmq.broker.processor.ReplyMessageProcessor;
 import org.apache.rocketmq.broker.processor.SendMessageProcessor;
 import org.apache.rocketmq.broker.slave.SlaveSynchronize;
+import org.apache.rocketmq.broker.subscription.LmqSubscriptionGroupManager;
 import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
+import org.apache.rocketmq.broker.topic.LmqTopicConfigManager;
 import org.apache.rocketmq.broker.topic.TopicConfigManager;
 import org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener;
 import org.apache.rocketmq.broker.transaction.TransactionalMessageCheckService;
@@ -106,6 +110,7 @@ import org.apache.rocketmq.store.config.MessageStoreConfig;
 import org.apache.rocketmq.store.dledger.DLedgerCommitLog;
 import org.apache.rocketmq.store.stats.BrokerStats;
 import org.apache.rocketmq.store.stats.BrokerStatsManager;
+import org.apache.rocketmq.store.stats.LmqBrokerStatsManager;

 public class BrokerController {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
@@ -180,10 +185,10 @@ public class BrokerController {
        this.nettyServerConfig = nettyServerConfig;
        this.nettyClientConfig = nettyClientConfig;
        this.messageStoreConfig = messageStoreConfig;
-        this.consumerOffsetManager = new ConsumerOffsetManager(this);
-        this.topicConfigManager = new TopicConfigManager(this);
+        this.consumerOffsetManager = messageStoreConfig.isEnableLmq() ? new LmqConsumerOffsetManager(this) : new ConsumerOffsetManager(this);
+        this.topicConfigManager = messageStoreConfig.isEnableLmq() ? new LmqTopicConfigManager(this) : new TopicConfigManager(this);
        this.pullMessageProcessor = new PullMessageProcessor(this);
-        this.pullRequestHoldService = new PullRequestHoldService(this);
+        this.pullRequestHoldService = messageStoreConfig.isEnableLmq() ? new LmqPullRequestHoldService(this) : new PullRequestHoldService(this);
        this.messageArrivingListener = new NotifyMessageArrivingListener(this.pullRequestHoldService);
        this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
        this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);
@@ -191,7 +196,7 @@ public class BrokerController {
        this.producerManager = new ProducerManager();
        this.clientHousekeepingService = new ClientHousekeepingService(this);
        this.broker2Client = new Broker2Client(this);
-        this.subscriptionGroupManager = new SubscriptionGroupManager(this);
+        this.subscriptionGroupManager = messageStoreConfig.isEnableLmq() ? new LmqSubscriptionGroupManager(this) : new SubscriptionGroupManager(this);
        this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);
        this.filterServerManager = new FilterServerManager(this);

@@ -207,7 +212,8 @@ public class BrokerController {
        this.heartbeatThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getHeartbeatThreadPoolQueueCapacity());
        this.endTransactionThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getEndTransactionPoolQueueCapacity());

-        this.brokerStatsManager = new BrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat());
+        this.brokerStatsManager = messageStoreConfig.isEnableLmq() ? new LmqBrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat()) : new BrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat());
+
        this.setStoreHost(new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), this.getNettyServerConfig().getListenPort()));

        this.brokerFastFailure = new BrokerFastFailure(this);
@@ -39,6 +39,10 @@ public class BrokerPathConfigHelper {
        return rootDir + File.separator + "config" + File.separator + "consumerOffset.json";
    }

+    public static String getLmqConsumerOffsetPath(final String rootDir) {
+        return rootDir + File.separator + "config" + File.separator + "lmqConsumerOffset.json";
+    }
+
    public static String getSubscriptionGroupPath(final String rootDir) {
        return rootDir + File.separator + "config" + File.separator + "subscriptionGroup.json";
    }
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.longpolling;
+
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+
+
+public class LmqPullRequestHoldService extends PullRequestHoldService {
+    private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
+
+    public LmqPullRequestHoldService(BrokerController brokerController) {
+        super(brokerController);
+    }
+
+    @Override
+    public String getServiceName() {
+        return LmqPullRequestHoldService.class.getSimpleName();
+    }
+
+    @Override
+    public void checkHoldRequest() {
+        for (String key : pullRequestTable.keySet()) {
+            int idx = key.lastIndexOf(TOPIC_QUEUEID_SEPARATOR);
+            if (idx <= 0 || idx >= key.length() - 1) {
+                pullRequestTable.remove(key);
+                continue;
+            }
+            String topic = key.substring(0, idx);
+            int queueId = Integer.parseInt(key.substring(idx + 1));
+            final long offset = brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
+            try {
+                this.notifyMessageArriving(topic, queueId, offset);
+            } catch (Throwable e) {
+                LOGGER.error("check hold request failed. topic={}, queueId={}", topic, queueId, e);
+            }
+            if (MixAll.isLmq(topic)) {
+                ManyPullRequest mpr = pullRequestTable.get(key);
+                if (mpr == null || mpr.getPullRequestList() == null || mpr.getPullRequestList().isEmpty()) {
+                    pullRequestTable.remove(key);
+                }
+            }
+        }
+    }
+}
@@ -39,4 +39,8 @@ public class ManyPullRequest {

        return null;
    }
+
+    public ArrayList<PullRequest> getPullRequestList() {
+        return pullRequestList;
+    }
 }
@@ -31,10 +31,10 @@ import org.apache.rocketmq.store.ConsumeQueueExt;

 public class PullRequestHoldService extends ServiceThread {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
-    private static final String TOPIC_QUEUEID_SEPARATOR = "@";
-    private final BrokerController brokerController;
+    protected static final String TOPIC_QUEUEID_SEPARATOR = "@";
+    protected final BrokerController brokerController;
    private final SystemClock systemClock = new SystemClock();
-    private ConcurrentMap<String/* topic@queueId */, ManyPullRequest> pullRequestTable =
+    protected ConcurrentMap<String/* topic@queueId */, ManyPullRequest> pullRequestTable =
        new ConcurrentHashMap<String, ManyPullRequest>(1024);

    public PullRequestHoldService(final BrokerController brokerController) {
@@ -93,7 +93,7 @@ public class PullRequestHoldService extends ServiceThread {
        return PullRequestHoldService.class.getSimpleName();
    }

-    private void checkHoldRequest() {
+    protected void checkHoldRequest() {
        for (String key : this.pullRequestTable.keySet()) {
            String[] kArray = key.split(TOPIC_QUEUEID_SEPARATOR);
            if (2 == kArray.length) {
@@ -35,12 +35,12 @@ import org.apache.rocketmq.remoting.protocol.RemotingSerializable;

 public class ConsumerOffsetManager extends ConfigManager {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
-    private static final String TOPIC_GROUP_SEPARATOR = "@";
+    protected static final String TOPIC_GROUP_SEPARATOR = "@";

-    private ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, Long>> offsetTable =
+    protected ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, Long>> offsetTable =
        new ConcurrentHashMap<String, ConcurrentMap<Integer, Long>>(512);

-    private transient BrokerController brokerController;
+    protected transient BrokerController brokerController;

    public ConsumerOffsetManager() {
    }
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.offset;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.BrokerPathConfigHelper;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+public class LmqConsumerOffsetManager extends ConsumerOffsetManager {
+    private ConcurrentHashMap<String, Long> lmqOffsetTable = new ConcurrentHashMap<>(512);
+
+    public LmqConsumerOffsetManager(BrokerController brokerController) {
+        super(brokerController);
+    }
+
+    @Override
+    public long queryOffset(final String group, final String topic, final int queueId) {
+        if (!MixAll.isLmq(group)) {
+            return super.queryOffset(group, topic, queueId);
+        }
+        // topic@group
+        String key = topic + TOPIC_GROUP_SEPARATOR + group;
+        Long offset = lmqOffsetTable.get(key);
+        if (offset != null) {
+            return offset;
+        }
+        return -1;
+    }
+
+    @Override
+    public Map<Integer, Long> queryOffset(final String group, final String topic) {
+        if (!MixAll.isLmq(group)) {
+            return super.queryOffset(group, topic);
+        }
+        Map<Integer, Long> map = new HashMap<>();
+        // topic@group
+        String key = topic + TOPIC_GROUP_SEPARATOR + group;
+        Long offset = lmqOffsetTable.get(key);
+        if (offset != null) {
+            map.put(0, offset);
+        }
+        return map;
+    }
+
+    @Override
+    public void commitOffset(final String clientHost, final String group, final String topic, final int queueId,
+        final long offset) {
+        if (!MixAll.isLmq(group)) {
+            super.commitOffset(clientHost, group, topic, queueId, offset);
+            return;
+        }
+        // topic@group
+        String key = topic + TOPIC_GROUP_SEPARATOR + group;
+        lmqOffsetTable.put(key, offset);
+    }
+
+    @Override
+    public String encode() {
+        return this.encode(false);
+    }
+
+    @Override
+    public String configFilePath() {
+        return BrokerPathConfigHelper.getLmqConsumerOffsetPath(brokerController.getMessageStoreConfig().getStorePathRootDir());
+    }
+
+    @Override
+    public void decode(String jsonString) {
+        if (jsonString != null) {
+            LmqConsumerOffsetManager obj = RemotingSerializable.fromJson(jsonString, LmqConsumerOffsetManager.class);
+            if (obj != null) {
+                super.offsetTable = obj.offsetTable;
+                this.lmqOffsetTable = obj.lmqOffsetTable;
+            }
+        }
+    }
+
+    @Override
+    public String encode(final boolean prettyFormat) {
+        return RemotingSerializable.toJson(this, prettyFormat);
+    }
+
+    public ConcurrentHashMap<String, Long> getLmqOffsetTable() {
+        return lmqOffsetTable;
+    }
+
+    public void setLmqOffsetTable(ConcurrentHashMap<String, Long> lmqOffsetTable) {
+        this.lmqOffsetTable = lmqOffsetTable;
+    }
+}
@@ -1,267 +1,272 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.rocketmq.broker.plugin;
-
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.common.message.MessageExtBatch;
-import org.apache.rocketmq.store.CommitLogDispatcher;
-import org.apache.rocketmq.store.ConsumeQueue;
-import org.apache.rocketmq.store.GetMessageResult;
-import org.apache.rocketmq.store.MessageExtBrokerInner;
-import org.apache.rocketmq.store.MessageFilter;
-import org.apache.rocketmq.store.MessageStore;
-import org.apache.rocketmq.store.PutMessageResult;
-import org.apache.rocketmq.store.QueryMessageResult;
-import org.apache.rocketmq.store.SelectMappedBufferResult;
-import org.apache.rocketmq.store.stats.BrokerStatsManager;
-
-public abstract class AbstractPluginMessageStore implements MessageStore {
-    protected MessageStore next = null;
-    protected MessageStorePluginContext context;
-
-    public AbstractPluginMessageStore(MessageStorePluginContext context, MessageStore next) {
-        this.next = next;
-        this.context = context;
-    }
-
-    @Override
-    public long getEarliestMessageTime() {
-        return next.getEarliestMessageTime();
-    }
-
-    @Override
-    public long lockTimeMills() {
-        return next.lockTimeMills();
-    }
-
-    @Override
-    public boolean isOSPageCacheBusy() {
-        return next.isOSPageCacheBusy();
-    }
-
-    @Override
-    public boolean isTransientStorePoolDeficient() {
-        return next.isTransientStorePoolDeficient();
-    }
-
-    @Override
-    public boolean load() {
-        return next.load();
-    }
-
-    @Override
-    public void start() throws Exception {
-        next.start();
-    }
-
-    @Override
-    public void shutdown() {
-        next.shutdown();
-    }
-
-    @Override
-    public void destroy() {
-        next.destroy();
-    }
-
-    @Override
-    public PutMessageResult putMessage(MessageExtBrokerInner msg) {
-        return next.putMessage(msg);
-    }
-
-    @Override
-    public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {
-        return next.asyncPutMessage(msg);
-    }
-
-    @Override
-    public CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch messageExtBatch) {
-        return next.asyncPutMessages(messageExtBatch);
-    }
-
-    @Override
-    public GetMessageResult getMessage(String group, String topic, int queueId, long offset,
-        int maxMsgNums, final MessageFilter messageFilter) {
-        return next.getMessage(group, topic, queueId, offset, maxMsgNums, messageFilter);
-    }
-
-    @Override
-    public long getMaxOffsetInQueue(String topic, int queueId) {
-        return next.getMaxOffsetInQueue(topic, queueId);
-    }
-
-    @Override
-    public long getMinOffsetInQueue(String topic, int queueId) {
-        return next.getMinOffsetInQueue(topic, queueId);
-    }
-
-    @Override
-    public long getCommitLogOffsetInQueue(String topic, int queueId, long consumeQueueOffset) {
-        return next.getCommitLogOffsetInQueue(topic, queueId, consumeQueueOffset);
-    }
-
-    @Override
-    public long getOffsetInQueueByTime(String topic, int queueId, long timestamp) {
-        return next.getOffsetInQueueByTime(topic, queueId, timestamp);
-    }
-
-    @Override
-    public MessageExt lookMessageByOffset(long commitLogOffset) {
-        return next.lookMessageByOffset(commitLogOffset);
-    }
-
-    @Override
-    public SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset) {
-        return next.selectOneMessageByOffset(commitLogOffset);
-    }
-
-    @Override
-    public SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset, int msgSize) {
-        return next.selectOneMessageByOffset(commitLogOffset, msgSize);
-    }
-
-    @Override
-    public String getRunningDataInfo() {
-        return next.getRunningDataInfo();
-    }
-
-    @Override
-    public HashMap<String, String> getRuntimeInfo() {
-        return next.getRuntimeInfo();
-    }
-
-    @Override
-    public long getMaxPhyOffset() {
-        return next.getMaxPhyOffset();
-    }
-
-    @Override
-    public long getMinPhyOffset() {
-        return next.getMinPhyOffset();
-    }
-
-    @Override
-    public long getEarliestMessageTime(String topic, int queueId) {
-        return next.getEarliestMessageTime(topic, queueId);
-    }
-
-    @Override
-    public long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueueOffset) {
-        return next.getMessageStoreTimeStamp(topic, queueId, consumeQueueOffset);
-    }
-
-    @Override
-    public long getMessageTotalInQueue(String topic, int queueId) {
-        return next.getMessageTotalInQueue(topic, queueId);
-    }
-
-    @Override
-    public SelectMappedBufferResult getCommitLogData(long offset) {
-        return next.getCommitLogData(offset);
-    }
-
-    @Override
-    public boolean appendToCommitLog(long startOffset, byte[] data, int dataStart, int dataLength) {
-        return next.appendToCommitLog(startOffset, data, dataStart, dataLength);
-    }
-
-    @Override
-    public void executeDeleteFilesManually() {
-        next.executeDeleteFilesManually();
-    }
-
-    @Override
-    public QueryMessageResult queryMessage(String topic, String key, int maxNum, long begin,
-        long end) {
-        return next.queryMessage(topic, key, maxNum, begin, end);
-    }
-
-    @Override
-    public void updateHaMasterAddress(String newAddr) {
-        next.updateHaMasterAddress(newAddr);
-    }
-
-    @Override
-    public long slaveFallBehindMuch() {
-        return next.slaveFallBehindMuch();
-    }
-
-    @Override
-    public long now() {
-        return next.now();
-    }
-
-    @Override
-    public int cleanUnusedTopic(Set<String> topics) {
-        return next.cleanUnusedTopic(topics);
-    }
-
-    @Override
-    public void cleanExpiredConsumerQueue() {
-        next.cleanExpiredConsumerQueue();
-    }
-
-    @Override
-    public boolean checkInDiskByConsumeOffset(String topic, int queueId, long consumeOffset) {
-        return next.checkInDiskByConsumeOffset(topic, queueId, consumeOffset);
-    }
-
-    @Override
-    public long dispatchBehindBytes() {
-        return next.dispatchBehindBytes();
-    }
-
-    @Override
-    public long flush() {
-        return next.flush();
-    }
-
-    @Override
-    public boolean resetWriteOffset(long phyOffset) {
-        return next.resetWriteOffset(phyOffset);
-    }
-
-    @Override
-    public long getConfirmOffset() {
-        return next.getConfirmOffset();
-    }
-
-    @Override
-    public void setConfirmOffset(long phyOffset) {
-        next.setConfirmOffset(phyOffset);
-    }
-
-    @Override
-    public LinkedList<CommitLogDispatcher> getDispatcherList() {
-        return next.getDispatcherList();
-    }
-
-    @Override
-    public ConsumeQueue getConsumeQueue(String topic, int queueId) {
-        return next.getConsumeQueue(topic, queueId);
-    }
-
-    @Override
-    public BrokerStatsManager getBrokerStatsManager() {
-        return next.getBrokerStatsManager();
-    };
-}
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.broker.plugin;
+
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageExtBatch;
+import org.apache.rocketmq.store.CommitLogDispatcher;
+import org.apache.rocketmq.store.ConsumeQueue;
+import org.apache.rocketmq.store.GetMessageResult;
+import org.apache.rocketmq.store.MessageExtBrokerInner;
+import org.apache.rocketmq.store.MessageFilter;
+import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.PutMessageResult;
+import org.apache.rocketmq.store.QueryMessageResult;
+import org.apache.rocketmq.store.SelectMappedBufferResult;
+import org.apache.rocketmq.store.stats.BrokerStatsManager;
+
+public abstract class AbstractPluginMessageStore implements MessageStore {
+    protected MessageStore next = null;
+    protected MessageStorePluginContext context;
+
+    public AbstractPluginMessageStore(MessageStorePluginContext context, MessageStore next) {
+        this.next = next;
+        this.context = context;
+    }
+
+    @Override
+    public long getEarliestMessageTime() {
+        return next.getEarliestMessageTime();
+    }
+
+    @Override
+    public long lockTimeMills() {
+        return next.lockTimeMills();
+    }
+
+    @Override
+    public boolean isOSPageCacheBusy() {
+        return next.isOSPageCacheBusy();
+    }
+
+    @Override
+    public boolean isTransientStorePoolDeficient() {
+        return next.isTransientStorePoolDeficient();
+    }
+
+    @Override
+    public boolean load() {
+        return next.load();
+    }
+
+    @Override
+    public void start() throws Exception {
+        next.start();
+    }
+
+    @Override
+    public void shutdown() {
+        next.shutdown();
+    }
+
+    @Override
+    public void destroy() {
+        next.destroy();
+    }
+
+    @Override
+    public PutMessageResult putMessage(MessageExtBrokerInner msg) {
+        return next.putMessage(msg);
+    }
+
+    @Override
+    public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {
+        return next.asyncPutMessage(msg);
+    }
+
+    @Override
+    public CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch messageExtBatch) {
+        return next.asyncPutMessages(messageExtBatch);
+    }
+
+    @Override
+    public GetMessageResult getMessage(String group, String topic, int queueId, long offset,
+        int maxMsgNums, final MessageFilter messageFilter) {
+        return next.getMessage(group, topic, queueId, offset, maxMsgNums, messageFilter);
+    }
+
+    @Override
+    public long getMaxOffsetInQueue(String topic, int queueId) {
+        return next.getMaxOffsetInQueue(topic, queueId);
+    }
+
+    @Override
+    public long getMinOffsetInQueue(String topic, int queueId) {
+        return next.getMinOffsetInQueue(topic, queueId);
+    }
+
+    @Override
+    public long getCommitLogOffsetInQueue(String topic, int queueId, long consumeQueueOffset) {
+        return next.getCommitLogOffsetInQueue(topic, queueId, consumeQueueOffset);
+    }
+
+    @Override
+    public long getOffsetInQueueByTime(String topic, int queueId, long timestamp) {
+        return next.getOffsetInQueueByTime(topic, queueId, timestamp);
+    }
+
+    @Override
+    public MessageExt lookMessageByOffset(long commitLogOffset) {
+        return next.lookMessageByOffset(commitLogOffset);
+    }
+
+    @Override
+    public SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset) {
+        return next.selectOneMessageByOffset(commitLogOffset);
+    }
+
+    @Override
+    public SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset, int msgSize) {
+        return next.selectOneMessageByOffset(commitLogOffset, msgSize);
+    }
+
+    @Override
+    public String getRunningDataInfo() {
+        return next.getRunningDataInfo();
+    }
+
+    @Override
+    public HashMap<String, String> getRuntimeInfo() {
+        return next.getRuntimeInfo();
+    }
+
+    @Override
+    public long getMaxPhyOffset() {
+        return next.getMaxPhyOffset();
+    }
+
+    @Override
+    public long getMinPhyOffset() {
+        return next.getMinPhyOffset();
+    }
+
+    @Override
+    public long getEarliestMessageTime(String topic, int queueId) {
+        return next.getEarliestMessageTime(topic, queueId);
+    }
+
+    @Override
+    public long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueueOffset) {
+        return next.getMessageStoreTimeStamp(topic, queueId, consumeQueueOffset);
+    }
+
+    @Override
+    public long getMessageTotalInQueue(String topic, int queueId) {
+        return next.getMessageTotalInQueue(topic, queueId);
+    }
+
+    @Override
+    public SelectMappedBufferResult getCommitLogData(long offset) {
+        return next.getCommitLogData(offset);
+    }
+
+    @Override
+    public boolean appendToCommitLog(long startOffset, byte[] data, int dataStart, int dataLength) {
+        return next.appendToCommitLog(startOffset, data, dataStart, dataLength);
+    }
+
+    @Override
+    public void executeDeleteFilesManually() {
+        next.executeDeleteFilesManually();
+    }
+
+    @Override
+    public QueryMessageResult queryMessage(String topic, String key, int maxNum, long begin,
+        long end) {
+        return next.queryMessage(topic, key, maxNum, begin, end);
+    }
+
+    @Override
+    public void updateHaMasterAddress(String newAddr) {
+        next.updateHaMasterAddress(newAddr);
+    }
+
+    @Override
+    public long slaveFallBehindMuch() {
+        return next.slaveFallBehindMuch();
+    }
+
+    @Override
+    public long now() {
+        return next.now();
+    }
+
+    @Override
+    public int cleanUnusedTopic(Set<String> topics) {
+        return next.cleanUnusedTopic(topics);
+    }
+
+    @Override
+    public void cleanExpiredConsumerQueue() {
+        next.cleanExpiredConsumerQueue();
+    }
+
+    @Override
+    public boolean checkInDiskByConsumeOffset(String topic, int queueId, long consumeOffset) {
+        return next.checkInDiskByConsumeOffset(topic, queueId, consumeOffset);
+    }
+
+    @Override
+    public long dispatchBehindBytes() {
+        return next.dispatchBehindBytes();
+    }
+
+    @Override
+    public long flush() {
+        return next.flush();
+    }
+
+    @Override
+    public boolean resetWriteOffset(long phyOffset) {
+        return next.resetWriteOffset(phyOffset);
+    }
+
+    @Override
+    public long getConfirmOffset() {
+        return next.getConfirmOffset();
+    }
+
+    @Override
+    public void setConfirmOffset(long phyOffset) {
+        next.setConfirmOffset(phyOffset);
+    }
+
+    @Override
+    public LinkedList<CommitLogDispatcher> getDispatcherList() {
+        return next.getDispatcherList();
+    }
+
+    @Override
+    public ConsumeQueue getConsumeQueue(String topic, int queueId) {
+        return next.getConsumeQueue(topic, queueId);
+    }
+
+    @Override
+    public BrokerStatsManager getBrokerStatsManager() {
+        return next.getBrokerStatsManager();
+    }
+
+    @Override
+    public void cleanUnusedLmqTopic(String topic) {
+        next.cleanUnusedLmqTopic(topic);
+    }
+}
@@ -292,6 +292,13 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
            return response;
        }

+        if (MixAll.isLmq(topic)) {
+            this.brokerController.getMessageStore().cleanUnusedLmqTopic(topic);
+            response.setCode(ResponseCode.SUCCESS);
+            response.setRemark(null);
+            return response;
+        }
+
        this.brokerController.getTopicConfigManager().deleteTopicConfig(topic);
        this.brokerController.getMessageStore()
            .cleanUnusedTopic(this.brokerController.getTopicConfigManager().getTopicConfigTable().keySet());
@@ -514,6 +514,10 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("[PC_SYNCHRONIZED]broker busy, start flow control for a while");
                break;
+            case LMQ_CONSUME_QUEUE_NUM_EXCEEDED:
+                response.setCode(ResponseCode.SYSTEM_ERROR);
+                response.setRemark("[LMQ_CONSUME_QUEUE_NUM_EXCEEDED]broker config enableLmq and enableMultiDispatch, lmq consumeQueue num exceed maxLmqConsumeQueueNum config num, default limit 2w.");
+                break;
            case UNKNOWN_ERROR:
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("UNKNOWN_ERROR");
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.subscription;
+
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+
+public class LmqSubscriptionGroupManager extends SubscriptionGroupManager {
+
+    public LmqSubscriptionGroupManager(BrokerController brokerController) {
+        super(brokerController);
+    }
+
+    @Override
+    public SubscriptionGroupConfig findSubscriptionGroupConfig(final String group) {
+        if (MixAll.isLmq(group)) {
+            SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
+            subscriptionGroupConfig.setGroupName(group);
+            return subscriptionGroupConfig;
+        }
+        return super.findSubscriptionGroupConfig(group);
+    }
+
+    @Override
+    public void updateSubscriptionGroupConfig(final SubscriptionGroupConfig config) {
+        if (config == null || MixAll.isLmq(config.getGroupName())) {
+            return;
+        }
+        super.updateSubscriptionGroupConfig(config);
+    }
+}
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.topic;
+
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.constant.PermName;
+
+public class LmqTopicConfigManager extends TopicConfigManager {
+    public LmqTopicConfigManager(BrokerController brokerController) {
+        super(brokerController);
+    }
+
+    @Override
+    public TopicConfig selectTopicConfig(final String topic) {
+        if (MixAll.isLmq(topic)) {
+            return simpleLmqTopicConfig(topic);
+        }
+        return super.selectTopicConfig(topic);
+    }
+
+    @Override
+    public void updateTopicConfig(final TopicConfig topicConfig) {
+        if (topicConfig == null || MixAll.isLmq(topicConfig.getTopicName())) {
+            return;
+        }
+        super.updateTopicConfig(topicConfig);
+    }
+
+    private TopicConfig simpleLmqTopicConfig(String topic) {
+        return new TopicConfig(topic, 1, 1, PermName.PERM_READ | PermName.PERM_WRITE);
+    }
+
+}
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.broker;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class BrokerPathConfigHelperTest {
+
+    @Test
+    public void testGetLmqConsumerOffsetPath() {
+        String lmqConsumerOffsetPath = BrokerPathConfigHelper.getLmqConsumerOffsetPath("/home/admin/store");
+        assertEquals("/home/admin/store/config/lmqConsumerOffset.json", lmqConsumerOffsetPath);
+
+
+        String consumerOffsetPath = BrokerPathConfigHelper.getConsumerOffsetPath("/home/admin/store");
+        assertEquals("/home/admin/store/config/consumerOffset.json", consumerOffsetPath);
+
+        String topicConfigPath = BrokerPathConfigHelper.getTopicConfigPath("/home/admin/store");
+        assertEquals("/home/admin/store/config/topics.json", topicConfigPath);
+
+        String subscriptionGroupPath = BrokerPathConfigHelper.getSubscriptionGroupPath("/home/admin/store");
+        assertEquals("/home/admin/store/config/subscriptionGroup.json", subscriptionGroupPath);
+
+    }
+}
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.broker.offset;
+
+import java.io.File;
+import java.util.Map;
+
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.subscription.LmqSubscriptionGroupManager;
+import org.apache.rocketmq.broker.topic.LmqTopicConfigManager;
+import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.netty.NettyClientConfig;
+import org.apache.rocketmq.remoting.netty.NettyServerConfig;
+import org.apache.rocketmq.store.config.MessageStoreConfig;
+import org.junit.After;
+import org.junit.Test;
+import org.mockito.Spy;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class LmqConsumerOffsetManagerTest {
+
+    @Spy
+    private BrokerController brokerController = new BrokerController(new BrokerConfig(), new NettyServerConfig(),
+        new NettyClientConfig(), new MessageStoreConfig());
+
+    @Test
+    public void testOffsetManage() {
+        LmqConsumerOffsetManager lmqConsumerOffsetManager = new LmqConsumerOffsetManager(brokerController);
+        LmqTopicConfigManager lmqTopicConfigManager = new LmqTopicConfigManager(brokerController);
+        LmqSubscriptionGroupManager lmqSubscriptionGroupManager = new LmqSubscriptionGroupManager(brokerController);
+
+        String lmqTopicName = "%LMQ%1111";
+        TopicConfig topicConfig = new TopicConfig();
+        topicConfig.setTopicName(lmqTopicName);
+        lmqTopicConfigManager.updateTopicConfig(topicConfig);
+        TopicConfig topicConfig1 = lmqTopicConfigManager.selectTopicConfig(lmqTopicName);
+        assertThat(topicConfig1.getTopicName()).isEqualTo(topicConfig.getTopicName());
+
+        String lmqGroupName = "%LMQ%GID_test";
+        SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
+        subscriptionGroupConfig.setGroupName(lmqGroupName);
+        lmqSubscriptionGroupManager.updateSubscriptionGroupConfig(subscriptionGroupConfig);
+        SubscriptionGroupConfig subscriptionGroupConfig1 = lmqSubscriptionGroupManager.findSubscriptionGroupConfig(
+            lmqGroupName);
+        assertThat(subscriptionGroupConfig1.getGroupName()).isEqualTo(subscriptionGroupConfig.getGroupName());
+
+        lmqConsumerOffsetManager.commitOffset("127.0.0.1", lmqGroupName, lmqTopicName, 0, 10L);
+        Map<Integer, Long> integerLongMap = lmqConsumerOffsetManager.queryOffset(lmqGroupName, lmqTopicName);
+        assertThat(integerLongMap.get(0)).isEqualTo(10L);
+        long offset = lmqConsumerOffsetManager.queryOffset(lmqGroupName, lmqTopicName, 0);
+        assertThat(offset).isEqualTo(10L);
+
+        long offset1 = lmqConsumerOffsetManager.queryOffset(lmqGroupName, lmqTopicName + "test", 0);
+        assertThat(offset1).isEqualTo(-1L);
+    }
+
+    @After
+    public void destroy() {
+        UtilAll.deleteFile(new File(new MessageStoreConfig().getStorePathRootDir()));
+    }
+
+}
@@ -83,6 +83,8 @@ public class MixAll {
    public static final String CID_SYS_RMQ_TRANS = "CID_RMQ_SYS_TRANS";
    public static final String ACL_CONF_TOOLS_FILE = "/conf/tools.yml";
    public static final String REPLY_MESSAGE_FLAG = "reply";
+    public static final String LMQ_PREFIX = "%LMQ%";
+    public static final String MULTI_DISPATCH_QUEUE_SPLITTER = ",";
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);

    public static String getWSAddr() {
@@ -444,4 +446,7 @@ public class MixAll {
        return String.format("%.1f %sB", bytes / Math.pow(unit, exp), pre);
    }

+    public static boolean isLmq(String lmqMetaData) {
+        return lmqMetaData != null && lmqMetaData.startsWith(LMQ_PREFIX);
+    }
 }
@@ -52,6 +52,8 @@ public class MessageConst {
    public static final String PROPERTY_PUSH_REPLY_TIME = "PUSH_REPLY_TIME";
    public static final String PROPERTY_CLUSTER = "CLUSTER";
    public static final String PROPERTY_MESSAGE_TYPE = "MSG_TYPE";
+    public static final String PROPERTY_INNER_MULTI_DISPATCH = "INNER_MULTI_DISPATCH";
+    public static final String PROPERTY_INNER_MULTI_QUEUE_OFFSET = "INNER_MULTI_QUEUE_OFFSET";

    public static final String KEY_SEPARATOR = " ";

@@ -88,5 +90,6 @@ public class MessageConst {
        STRING_HASH_SET.add(PROPERTY_PUSH_REPLY_TIME);
        STRING_HASH_SET.add(PROPERTY_CLUSTER);
        STRING_HASH_SET.add(PROPERTY_MESSAGE_TYPE);
+        STRING_HASH_SET.add(PROPERTY_INNER_MULTI_QUEUE_OFFSET);
    }
 }
@@ -95,4 +95,16 @@ public class MixAllTest {
        assertThat(MixAll.LOCALHOST).isNotNull();
        assertThat(MixAll.getLocalhostByNetworkInterface()).isNotNull();
    }
+
+    @Test
+    public void testIsLmq() {
+        String testLmq = null;
+        assertThat(MixAll.isLmq(testLmq)).isFalse();
+        testLmq = "lmq";
+        assertThat(MixAll.isLmq(testLmq)).isFalse();
+        testLmq = "%LMQ%queue123";
+        assertThat(MixAll.isLmq(testLmq)).isTrue();
+        testLmq = "%LMQ%GID_TEST";
+        assertThat(MixAll.isLmq(testLmq)).isTrue();
+    }
 }
@@ -0,0 +1,85 @@
+# Light message queue (LMQ)
+LMQ采用的读放大的策略，写一份数据，多个LMQ队列分发，
+因为存储的成本和效率对用户的体感最明显。写多份不仅加大了存储成本，同时也对性能和数据准确一致性提出了挑战。
+
+![](image/LMQ_1.png)
+
+上图描述的是LMQ的队列存储模型，消息可以来自各个接入场景
+（如服务端的MQ/AMQP，客户端的MQTT），但只会写一份存到commitlog里面，然后分发出多个需求场景的队列索引（ConsumerQueue），如服务端场景（MQ/AMQP）可以按照一级Topic队列进行传统的服务端消费，客户端MQTT场景可以按照MQTT多级Topic(也即 LMQ)进行消费消息。
+
+## 一、broker启动配置
+
+
+broker.conf文件需要增加以下的配置项，开启LMQ开关，这样就可以识别LMQ相关属性的消息，进行原子分发消息到LMQ队列
+```properties
+enableLmq = true
+enableMultiDispatch = true
+```
+## 二、发送消息
+发送消息的时候通过设置 INNER_MULTI_DISPATCH 属性，LMQ queue使用逗号分割，queue前缀必须是 %LMQ%，这样broker就可以识别LMQ queue.
+以下代码只是demo伪代码 具体逻辑参照执行即可
+```java
+DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
+producer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
+producer.start();
+
+
+/*
+* Create a message instance, specifying topic, tag and message body.
+*/
+Message msg = new Message("TopicTest" /* Topic */,
+                          "TagA" /* Tag */,
+                          ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
+                         );
+/*
+* INNER_MULTI_DISPATCH property and PREFIX must start as "%LMQ%",
+* If it is multiple LMQ, need to use “,” split
+*/
+message.putUserProperty("INNER_MULTI_DISPATCH", "%LMQ%123,%LMQ%456");
+/*
+* Call send message to deliver message to one of brokers.
+*/
+SendResult sendResult = producer.send(msg);
+```
+## 三、拉取消息
+LMQ queue在每个broker上只有一个queue，也即queueId为0， 指明轻量级的MessageQueue，就可以拉取消息进行消费。
+以下代码只是demo伪代码 具体逻辑参照执行即可
+```java
+DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer();
+defaultMQPullConsumer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
+defaultMQPullConsumer.setVipChannelEnabled(false);
+defaultMQPullConsumer.setConsumerGroup("CID_RMQ_SYS_LMQ_TEST");
+defaultMQPullConsumer.setInstanceName("CID_RMQ_SYS_LMQ_TEST");
+defaultMQPullConsumer.setRegisterTopics(new HashSet<>(Arrays.asList("TopicTest")));
+defaultMQPullConsumer.setBrokerSuspendMaxTimeMillis(2000);
+defaultMQPullConsumer.setConsumerTimeoutMillisWhenSuspend(3000);
+defaultMQPullConsumer.start();
+
+String brokerName = "set broker Name";
+MessageQueue mq = new MessageQueue("%LMQ%123", brokerName, 0);
+defaultMQPullConsumer.getDefaultMQPullConsumerImpl().getRebalanceImpl().getmQClientFactory().updateTopicRouteInfoFromNameServer("TopicTest");
+
+Thread.sleep(30000);
+Long offset = defaultMQPullConsumer.maxOffset(mq);
+
+defaultMQPullConsumer.pullBlockIfNotFound(
+                mq, "*", offset, 32,
+                new PullCallback() {
+                    @Override
+                    public void onSuccess(PullResult pullResult) {
+                        List<MessageExt> list = pullResult.getMsgFoundList();
+                        if (list == null || list.isEmpty()) {
+                            return;
+                        }
+                        for (MessageExt messageExt : list) {
+                            System.out.println(messageExt);
+                        }    
+                    }
+                    @Override
+                    public void onException(Throwable e) {
+                       
+                    }
+});
+```
+
+
@@ -29,6 +29,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;

 import org.apache.rocketmq.common.ServiceThread;
@@ -68,6 +69,7 @@ public class CommitLog {
    private final AppendMessageCallback appendMessageCallback;
    private final ThreadLocal<PutMessageThreadLocal> putMessageThreadLocal;
    protected HashMap<String/* topic-queueid */, Long/* offset */> topicQueueTable = new HashMap<String, Long>(1024);
+    protected Map<String/* topic-queueid */, Long/* offset */> lmqTopicQueueTable = new ConcurrentHashMap<>(1024);
    protected volatile long confirmOffset = -1L;

    private volatile long beginTimeInLock = 0;
@@ -76,6 +78,8 @@ public class CommitLog {

    private volatile Set<String> fullStorePaths = Collections.emptySet();

+    private final MultiDispatch multiDispatch;
+
    public CommitLog(final DefaultMessageStore defaultMessageStore) {
        String storePath = defaultMessageStore.getMessageStoreConfig().getStorePathCommitLog();
        if (storePath.contains(MessageStoreConfig.MULTI_PATH_SPLITTER)) {
@@ -107,6 +111,8 @@ public class CommitLog {
        };
        this.putMessageLock = defaultMessageStore.getMessageStoreConfig().isUseReentrantLockWhenPutMessage() ? new PutMessageReentrantLock() : new PutMessageSpinLock();

+        this.multiDispatch = new MultiDispatch(defaultMessageStore, this);
+
    }

    public void setFullStorePaths(Set<String> fullStorePaths) {
@@ -117,6 +123,10 @@ public class CommitLog {
        return fullStorePaths;
    }

+    public ThreadLocal<PutMessageThreadLocal> getPutMessageThreadLocal() {
+        return putMessageThreadLocal;
+    }
+
    public boolean load() {
        boolean result = this.mappedFileQueue.load();
        log.info("load commit log " + (result ? "OK" : "Failed"));
@@ -964,6 +974,7 @@ public class CommitLog {
        String key = topic + "-" + queueId;
        synchronized (this) {
            this.topicQueueTable.remove(key);
+            this.lmqTopicQueueTable.remove(key);
        }

        log.info("removeQueueFromTopicQueueTable OK Topic: {} QueueId: {}", topic, queueId);
@@ -987,6 +998,10 @@ public class CommitLog {
        return diff;
    }

+    public Map<String, Long> getLmqTopicQueueTable() {
+        return this.lmqTopicQueueTable;
+    }
+
    abstract class FlushCommitLogService extends ServiceThread {
        protected static final int RETRY_TIMES_OVER = 10;
    }
@@ -1298,6 +1313,11 @@ public class CommitLog {
                CommitLog.this.topicQueueTable.put(key, queueOffset);
            }

+            boolean multiDispatchWrapResult = CommitLog.this.multiDispatch.wrapMultiDispatch(msgInner);
+            if (!multiDispatchWrapResult) {
+                return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
+            }
+
            // Transaction messages that require special handling
            final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
            switch (tranType) {
@@ -1361,6 +1381,7 @@ public class CommitLog {
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                    // The next update ConsumeQueue information
                    CommitLog.this.topicQueueTable.put(key, ++queueOffset);
+                    CommitLog.this.multiDispatch.updateMultiQueueOffset(msgInner);
                    break;
                default:
                    break;
@@ -1691,6 +1712,9 @@ public class CommitLog {
            byteBuffer.limit(limit);
        }

+        public ByteBuffer getEncoderBuffer() {
+            return encoderBuffer;
+        }
    }

    static class PutMessageThreadLocal {
@@ -19,7 +19,11 @@ package org.apache.rocketmq.store;
 import java.io.File;
 import java.nio.ByteBuffer;
 import java.util.List;
+import java.util.Map;
+
+import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.store.config.BrokerRole;
@@ -376,7 +380,7 @@ public class ConsumeQueue {
        return this.minLogicOffset / CQ_STORE_UNIT_SIZE;
    }

-    public void putMessagePositionInfoWrapper(DispatchRequest request) {
+    public void putMessagePositionInfoWrapper(DispatchRequest request, boolean multiQueue) {
        final int maxRetries = 30;
        boolean canWrite = this.defaultMessageStore.getRunningFlags().isCQWriteable();
        for (int i = 0; i < maxRetries && canWrite; i++) {
@@ -403,6 +407,9 @@ public class ConsumeQueue {
                    this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(request.getStoreTimestamp());
                }
                this.defaultMessageStore.getStoreCheckpoint().setLogicsMsgTimestamp(request.getStoreTimestamp());
+                if (multiQueue) {
+                    multiDispatchLmqQueue(request, maxRetries);
+                }
                return;
            } else {
                // XXX: warn and notify me
@@ -422,6 +429,52 @@ public class ConsumeQueue {
        this.defaultMessageStore.getRunningFlags().makeLogicsQueueError();
    }

+    private void multiDispatchLmqQueue(DispatchRequest request, int maxRetries) {
+        Map<String, String> prop = request.getPropertiesMap();
+        String multiDispatchQueue = prop.get(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
+        String multiQueueOffset = prop.get(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET);
+        String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
+        String[] queueOffsets = multiQueueOffset.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
+        if (queues.length != queueOffsets.length) {
+            log.error("[bug] queues.length!=queueOffsets.length ", request.getTopic());
+            return;
+        }
+        for (int i = 0; i < queues.length; i++) {
+            String queueName = queues[i];
+            long queueOffset = Long.parseLong(queueOffsets[i]);
+            int queueId = request.getQueueId();
+            if (this.defaultMessageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(queueName)) {
+                queueId = 0;
+            }
+            doDispatchLmqQueue(request, maxRetries, queueName, queueOffset, queueId);
+
+        }
+        return;
+    }
+
+    private void doDispatchLmqQueue(DispatchRequest request, int maxRetries, String queueName, long queueOffset,
+        int queueId) {
+        ConsumeQueue cq = this.defaultMessageStore.findConsumeQueue(queueName, queueId);
+        boolean canWrite = this.defaultMessageStore.getRunningFlags().isCQWriteable();
+        for (int i = 0; i < maxRetries && canWrite; i++) {
+            boolean result = cq.putMessagePositionInfo(request.getCommitLogOffset(), request.getMsgSize(),
+                request.getTagsCode(),
+                queueOffset);
+            if (result) {
+                break;
+            } else {
+                log.warn("[BUG]put commit log position info to " + queueName + ":" + queueId + " " + request.getCommitLogOffset()
+                    + " failed, retry " + i + " times");
+
+                try {
+                    Thread.sleep(1000);
+                } catch (InterruptedException e) {
+                    log.warn("", e);
+                }
+            }
+        }
+    }
+
    private boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,
        final long cqOffset) {

@@ -584,4 +637,5 @@ public class ConsumeQueue {
    public boolean isExtAddr(long tagsCode) {
        return ConsumeQueueExt.isExtAddr(tagsCode);
    }
+
 }
@@ -39,7 +39,10 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.ServiceThread;
@@ -47,6 +50,7 @@ import org.apache.rocketmq.common.SystemClock;
 import org.apache.rocketmq.common.ThreadFactoryImpl;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageDecoder;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageExtBatch;
@@ -109,6 +113,8 @@ public class DefaultMessageStore implements MessageStore {

    private AtomicLong printTimes = new AtomicLong(0);

+    private final AtomicInteger lmqConsumeQueueNum = new AtomicInteger(0);
+
    private final LinkedList<CommitLogDispatcher> dispatcherList;

    private RandomAccessFile lockFile;
@@ -419,6 +425,23 @@ public class DefaultMessageStore implements MessageStore {
        return PutMessageStatus.PUT_OK;
    }

+    private PutMessageStatus checkLmqMessage(MessageExtBrokerInner msg) {
+        if (msg.getProperties() != null
+            && StringUtils.isNotBlank(msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH))
+            && this.isLmqConsumeQueueNumExceeded()) {
+            return PutMessageStatus.LMQ_CONSUME_QUEUE_NUM_EXCEEDED;
+        }
+        return PutMessageStatus.PUT_OK;
+    }
+
+    private boolean isLmqConsumeQueueNumExceeded() {
+        if (this.getMessageStoreConfig().isEnableLmq() && this.getMessageStoreConfig().isEnableMultiDispatch()
+            && this.lmqConsumeQueueNum.get() > this.messageStoreConfig.getMaxLmqConsumeQueueNum()) {
+            return true;
+        }
+        return false;
+    }
+
    @Override
    public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {
        PutMessageStatus checkStoreStatus = this.checkStoreStatus();
@@ -431,6 +454,12 @@ public class DefaultMessageStore implements MessageStore {
            return CompletableFuture.completedFuture(new PutMessageResult(msgCheckStatus, null));
        }

+        PutMessageStatus lmqMsgCheckStatus = this.checkLmqMessage(msg);
+        if (msgCheckStatus == PutMessageStatus.LMQ_CONSUME_QUEUE_NUM_EXCEEDED) {
+            return CompletableFuture.completedFuture(new PutMessageResult(lmqMsgCheckStatus, null));
+        }
+
+
        long beginTime = this.getSystemClock().now();
        CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);

@@ -448,7 +477,6 @@ public class DefaultMessageStore implements MessageStore {

        return putResultFuture;
    }
-
    public CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch messageExtBatch) {
        PutMessageStatus checkStoreStatus = this.checkStoreStatus();
        if (checkStoreStatus != PutMessageStatus.PUT_OK) {
@@ -532,6 +560,11 @@ public class DefaultMessageStore implements MessageStore {
            return null;
        }

+        if (MixAll.isLmq(topic) && this.isLmqConsumeQueueNumExceeded()) {
+            log.warn("message store is not available, broker config enableLmq and enableMultiDispatch, lmq consumeQueue num exceed maxLmqConsumeQueueNum config num");
+            return null;
+        }
+
        long beginTime = this.getSystemClock().now();

        GetMessageStatus status = GetMessageStatus.NO_MESSAGE_IN_QUEUE;
@@ -1016,7 +1049,8 @@ public class DefaultMessageStore implements MessageStore {
            String topic = next.getKey();

            if (!topics.contains(topic) && !topic.equals(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC)
-                    && !topic.equals(TopicValidator.RMQ_SYS_TRANS_OP_HALF_TOPIC)) {
+                    && !topic.equals(TopicValidator.RMQ_SYS_TRANS_OP_HALF_TOPIC)
+                    && !MixAll.isLmq(topic)) {
                ConcurrentMap<Integer, ConsumeQueue> queueTable = next.getValue();
                for (ConsumeQueue cq : queueTable.values()) {
                    cq.destroy();
@@ -1218,6 +1252,9 @@ public class DefaultMessageStore implements MessageStore {
            if (oldLogic != null) {
                logic = oldLogic;
            } else {
+                if (MixAll.isLmq(topic)) {
+                    lmqConsumeQueueNum.getAndIncrement();
+                }
                logic = newLogic;
            }
        }
@@ -1424,6 +1461,9 @@ public class DefaultMessageStore implements MessageStore {
            map = new ConcurrentHashMap<Integer/* queueId */, ConsumeQueue>();
            map.put(queueId, consumeQueue);
            this.consumeQueueTable.put(topic, map);
+            if (MixAll.isLmq(topic)) {
+                this.lmqConsumeQueueNum.getAndIncrement();
+            }
        } else {
            map.put(queueId, consumeQueue);
        }
@@ -1497,7 +1537,23 @@ public class DefaultMessageStore implements MessageStore {

    public void putMessagePositionInfo(DispatchRequest dispatchRequest) {
        ConsumeQueue cq = this.findConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId());
-        cq.putMessagePositionInfoWrapper(dispatchRequest);
+        cq.putMessagePositionInfoWrapper(dispatchRequest, checkMultiDispatchQueue(dispatchRequest));
+    }
+
+    private boolean checkMultiDispatchQueue(DispatchRequest dispatchRequest) {
+        if (!this.messageStoreConfig.isEnableMultiDispatch()) {
+            return false;
+        }
+        Map<String, String> prop = dispatchRequest.getPropertiesMap();
+        if (prop == null && prop.isEmpty()) {
+            return false;
+        }
+        String multiDispatchQueue = prop.get(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
+        String multiQueueOffset = prop.get(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET);
+        if (StringUtils.isBlank(multiDispatchQueue) || StringUtils.isBlank(multiQueueOffset)) {
+            return false;
+        }
+        return true;
    }

    @Override
@@ -1517,6 +1573,29 @@ public class DefaultMessageStore implements MessageStore {

    }

+    @Override
+    public void cleanUnusedLmqTopic(String topic) {
+        if (this.consumeQueueTable.containsKey(topic)) {
+            ConcurrentMap<Integer, ConsumeQueue> map = this.consumeQueueTable.get(topic);
+            if (map != null) {
+                ConsumeQueue cq = map.get(0);
+                cq.destroy();
+                log.info("cleanUnusedLmqTopic: {} {} ConsumeQueue cleaned",
+                    cq.getTopic(),
+                    cq.getQueueId()
+                );
+
+                this.commitLog.removeQueueFromTopicQueueTable(cq.getTopic(), cq.getQueueId());
+                this.lmqConsumeQueueNum.getAndDecrement();
+            }
+            this.consumeQueueTable.remove(topic);
+            if (this.brokerConfig.isAutoDeleteUnusedStats()) {
+                this.brokerStatsManager.onTopicDeleted(topic);
+            }
+            log.info("cleanUnusedLmqTopic: {},topic destroyed", topic);
+        }
+    }
+
    public int remainTransientStoreBufferNumbs() {
        return this.transientStorePool.availableBufferNums();
    }
@@ -1974,6 +2053,7 @@ public class DefaultMessageStore implements MessageStore {
                                            dispatchRequest.getQueueId(), dispatchRequest.getConsumeQueueOffset() + 1,
                                            dispatchRequest.getTagsCode(), dispatchRequest.getStoreTimestamp(),
                                            dispatchRequest.getBitMap(), dispatchRequest.getPropertiesMap());
+                                        notifyMessageArrive4MultiQueue(dispatchRequest);
                                    }

                                    this.reputFromOffset += size;
@@ -2016,6 +2096,34 @@ public class DefaultMessageStore implements MessageStore {
            }
        }

+        private void notifyMessageArrive4MultiQueue(DispatchRequest dispatchRequest) {
+            Map<String, String> prop = dispatchRequest.getPropertiesMap();
+            if (prop == null) {
+                return;
+            }
+            String multiDispatchQueue = prop.get(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
+            String multiQueueOffset = prop.get(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET);
+            if (StringUtils.isBlank(multiDispatchQueue) || StringUtils.isBlank(multiQueueOffset)) {
+                return;
+            }
+            String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
+            String[] queueOffsets = multiQueueOffset.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
+            if (queues.length != queueOffsets.length) {
+                return;
+            }
+            for (int i = 0; i < queues.length; i++) {
+                String queueName = queues[i];
+                long queueOffset = Long.parseLong(queueOffsets[i]);
+                int queueId = dispatchRequest.getQueueId();
+                if (DefaultMessageStore.this.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(queueName)) {
+                    queueId = 0;
+                }
+                DefaultMessageStore.this.messageArrivingListener.arriving(
+                    queueName, queueId, queueOffset + 1, dispatchRequest.getTagsCode(),
+                    dispatchRequest.getStoreTimestamp(), dispatchRequest.getBitMap(), dispatchRequest.getPropertiesMap());
+            }
+        }
+
        @Override
        public void run() {
            DefaultMessageStore.log.info(this.getServiceName() + " service started");
@@ -397,4 +397,14 @@ public interface MessageStore {
     * @param brokerRole
     */
    void handleScheduleMessageService(BrokerRole brokerRole);
+
+    /**
+     * Clean unused lmq topic.
+     * When calling to clean up the lmq topic,
+     * the lmq topic cannot be used to write messages at the same time,
+     * otherwise the messages of the cleaning lmq topic may be lost,
+     * please call this method with caution
+     * @param topic lmq topic
+     */
+    void cleanUnusedLmqTopic(String topic);
 }
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.store;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.store.CommitLog.MessageExtEncoder;
+
+/**
+ * not-thread-safe
+ */
+public class MultiDispatch {
+    private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
+    private final StringBuilder keyBuilder = new StringBuilder();
+    private final DefaultMessageStore messageStore;
+    private final CommitLog commitLog;
+
+    public MultiDispatch(DefaultMessageStore messageStore, CommitLog commitLog) {
+        this.messageStore = messageStore;
+        this.commitLog = commitLog;
+    }
+
+    public String queueKey(String queueName, MessageExtBrokerInner msgInner) {
+        keyBuilder.setLength(0);
+        keyBuilder.append(queueName);
+        keyBuilder.append('-');
+        int queueId = msgInner.getQueueId();
+        if (messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(queueName)) {
+            queueId = 0;
+        }
+        keyBuilder.append(queueId);
+        return keyBuilder.toString();
+    }
+
+    public boolean wrapMultiDispatch(final MessageExtBrokerInner msgInner) {
+        if (!messageStore.getMessageStoreConfig().isEnableMultiDispatch()) {
+            return true;
+        }
+        String multiDispatchQueue = msgInner.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
+        if (StringUtils.isBlank(multiDispatchQueue)) {
+            return true;
+        }
+        String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
+        Long[] queueOffsets = new Long[queues.length];
+        for (int i = 0; i < queues.length; i++) {
+            String key = queueKey(queues[i], msgInner);
+            Long queueOffset;
+            try {
+                queueOffset = getTopicQueueOffset(key);
+            } catch (Exception e) {
+                return false;
+            }
+            if (null == queueOffset) {
+                queueOffset = 0L;
+                if (messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(key)) {
+                    commitLog.getLmqTopicQueueTable().put(key, queueOffset);
+                } else {
+                    commitLog.getTopicQueueTable().put(key, queueOffset);
+                }
+            }
+            queueOffsets[i] = queueOffset;
+        }
+        MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET,
+            StringUtils.join(queueOffsets, MixAll.MULTI_DISPATCH_QUEUE_SPLITTER));
+        removeWaitStorePropertyString(msgInner);
+        return rebuildMsgInner(msgInner);
+    }
+
+    private void removeWaitStorePropertyString(MessageExtBrokerInner msgInner) {
+        if (msgInner.getProperties().containsKey(MessageConst.PROPERTY_WAIT_STORE_MSG_OK)) {
+            // There is no need to store "WAIT=true", remove it from propertiesString to save 9 bytes for each message.
+            // It works for most case. In some cases msgInner.setPropertiesString invoked later and replace it.
+            String waitStoreMsgOKValue = msgInner.getProperties().remove(MessageConst.PROPERTY_WAIT_STORE_MSG_OK);
+            msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
+            // Reput to properties, since msgInner.isWaitStoreMsgOK() will be invoked later
+            msgInner.getProperties().put(MessageConst.PROPERTY_WAIT_STORE_MSG_OK, waitStoreMsgOKValue);
+        } else {
+            msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
+        }
+    }
+
+    private boolean rebuildMsgInner(MessageExtBrokerInner msgInner) {
+        MessageExtEncoder encoder = this.commitLog.getPutMessageThreadLocal().get().getEncoder();
+        PutMessageResult encodeResult = encoder.encode(msgInner);
+        if (encodeResult != null) {
+            LOGGER.error("rebuild msgInner for multiDispatch", encodeResult);
+            return false;
+        }
+        msgInner.setEncodedBuff(encoder.getEncoderBuffer());
+        return true;
+
+    }
+
+    public void updateMultiQueueOffset(final MessageExtBrokerInner msgInner) {
+        if (!messageStore.getMessageStoreConfig().isEnableMultiDispatch()) {
+            return;
+        }
+        String multiDispatchQueue = msgInner.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
+        if (StringUtils.isBlank(multiDispatchQueue)) {
+            return;
+        }
+        String multiQueueOffset = msgInner.getProperty(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET);
+        if (StringUtils.isBlank(multiQueueOffset)) {
+            LOGGER.error("[bug] no multiQueueOffset when updating {}", msgInner.getTopic());
+            return;
+        }
+        String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
+        String[] queueOffsets = multiQueueOffset.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
+        if (queues.length != queueOffsets.length) {
+            LOGGER.error("[bug] num is not equal when updateMultiQueueOffset {}", msgInner.getTopic());
+            return;
+        }
+        for (int i = 0; i < queues.length; i++) {
+            String key = queueKey(queues[i], msgInner);
+            long queueOffset = Long.parseLong(queueOffsets[i]);
+            if (messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(key)) {
+                commitLog.getLmqTopicQueueTable().put(key, ++queueOffset);
+            } else {
+                commitLog.getTopicQueueTable().put(key, ++queueOffset);
+            }
+        }
+    }
+
+    private Long getTopicQueueOffset(String key) throws Exception {
+        Long offset = null;
+        if (messageStore.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(key)) {
+            Long queueNextOffset = commitLog.getLmqTopicQueueTable().get(key);
+            if (queueNextOffset != null) {
+                offset = queueNextOffset;
+            }
+        } else {
+            offset = commitLog.getTopicQueueTable().get(key);
+        }
+        return offset;
+    }
+
+}
@@ -27,4 +27,5 @@ public enum PutMessageStatus {
    PROPERTIES_SIZE_EXCEEDED,
    OS_PAGECACHE_BUSY,
    UNKNOWN_ERROR,
+    LMQ_CONSUME_QUEUE_NUM_EXCEEDED,
 }
@@ -159,6 +159,10 @@ public class MessageStoreConfig {

    private boolean enableScheduleMessageStats = true;

+    private boolean enableLmq = false;
+    private boolean enableMultiDispatch = false;
+    private int maxLmqConsumeQueueNum = 20000;
+
    public boolean isDebugLockEnable() {
        return debugLockEnable;
    }
@@ -744,4 +748,28 @@ public class MessageStoreConfig {
    public void setEnableScheduleMessageStats(boolean enableScheduleMessageStats) {
        this.enableScheduleMessageStats = enableScheduleMessageStats;
    }
+
+    public boolean isEnableLmq() {
+        return enableLmq;
+    }
+
+    public void setEnableLmq(boolean enableLmq) {
+        this.enableLmq = enableLmq;
+    }
+
+    public boolean isEnableMultiDispatch() {
+        return enableMultiDispatch;
+    }
+
+    public void setEnableMultiDispatch(boolean enableMultiDispatch) {
+        this.enableMultiDispatch = enableMultiDispatch;
+    }
+
+    public int getMaxLmqConsumeQueueNum() {
+        return maxLmqConsumeQueueNum;
+    }
+
+    public void setMaxLmqConsumeQueueNum(int maxLmqConsumeQueueNum) {
+        this.maxLmqConsumeQueueNum = maxLmqConsumeQueueNum;
+    }
 }
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.store.stats;
+
+import org.apache.rocketmq.common.MixAll;
+
+public class LmqBrokerStatsManager extends BrokerStatsManager {
+
+    public LmqBrokerStatsManager(String clusterName, boolean enableQueueStat) {
+        super(clusterName, enableQueueStat);
+    }
+
+    @Override
+    public void incGroupGetNums(final String group, final String topic, final int incValue) {
+        String lmqGroup = group;
+        String lmqTopic = topic;
+        if (MixAll.isLmq(group)) {
+            lmqGroup = MixAll.LMQ_PREFIX;
+        }
+        if (MixAll.isLmq(topic)) {
+            lmqTopic = MixAll.LMQ_PREFIX;
+        }
+        super.incGroupGetNums(lmqGroup, lmqTopic, incValue);
+    }
+
+    @Override
+    public void incGroupGetSize(final String group, final String topic, final int incValue) {
+        String lmqGroup = group;
+        String lmqTopic = topic;
+        if (MixAll.isLmq(group)) {
+            lmqGroup = MixAll.LMQ_PREFIX;
+        }
+        if (MixAll.isLmq(topic)) {
+            lmqTopic = MixAll.LMQ_PREFIX;
+        }
+        super.incGroupGetSize(lmqGroup, lmqTopic, incValue);
+    }
+
+    @Override
+    public void incGroupGetLatency(final String group, final String topic, final int queueId, final int incValue) {
+        String lmqGroup = group;
+        String lmqTopic = topic;
+        if (MixAll.isLmq(group)) {
+            lmqGroup = MixAll.LMQ_PREFIX;
+        }
+        if (MixAll.isLmq(topic)) {
+            lmqTopic = MixAll.LMQ_PREFIX;
+        }
+        super.incGroupGetLatency(lmqGroup, lmqTopic, queueId, incValue);
+    }
+
+    @Override
+    public void incSendBackNums(final String group, final String topic) {
+        String lmqGroup = group;
+        String lmqTopic = topic;
+        if (MixAll.isLmq(group)) {
+            lmqGroup = MixAll.LMQ_PREFIX;
+        }
+        if (MixAll.isLmq(topic)) {
+            lmqTopic = MixAll.LMQ_PREFIX;
+        }
+        super.incSendBackNums(lmqGroup, lmqTopic);
+    }
+
+    @Override
+    public double tpsGroupGetNums(final String group, final String topic) {
+        String lmqGroup = group;
+        String lmqTopic = topic;
+        if (MixAll.isLmq(group)) {
+            lmqGroup = MixAll.LMQ_PREFIX;
+        }
+        if (MixAll.isLmq(topic)) {
+            lmqTopic = MixAll.LMQ_PREFIX;
+        }
+        return super.tpsGroupGetNums(lmqGroup, lmqTopic);
+    }
+
+    @Override
+    public void recordDiskFallBehindTime(final String group, final String topic, final int queueId,
+        final long fallBehind) {
+        String lmqGroup = group;
+        String lmqTopic = topic;
+        if (MixAll.isLmq(group)) {
+            lmqGroup = MixAll.LMQ_PREFIX;
+        }
+        if (MixAll.isLmq(topic)) {
+            lmqTopic = MixAll.LMQ_PREFIX;
+        }
+        super.recordDiskFallBehindTime(lmqGroup, lmqTopic, queueId, fallBehind);
+    }
+
+    @Override
+    public void recordDiskFallBehindSize(final String group, final String topic, final int queueId,
+        final long fallBehind) {
+        String lmqGroup = group;
+        String lmqTopic = topic;
+        if (MixAll.isLmq(group)) {
+            lmqGroup = MixAll.LMQ_PREFIX;
+        }
+        if (MixAll.isLmq(topic)) {
+            lmqTopic = MixAll.LMQ_PREFIX;
+        }
+        super.recordDiskFallBehindSize(lmqGroup, lmqTopic, queueId, fallBehind);
+    }
+
+}
@@ -27,6 +27,7 @@ import java.nio.ByteBuffer;
 import java.util.Map;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageDecoder;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 import org.apache.rocketmq.store.stats.BrokerStatsManager;
@@ -146,6 +147,34 @@ public class ConsumeQueueTest {
        return master;
    }

+    protected DefaultMessageStore genForMultiQueue() throws Exception {
+        MessageStoreConfig messageStoreConfig = buildStoreConfig(
+            commitLogFileSize, cqFileSize, true, cqExtFileSize
+        );
+
+        messageStoreConfig.setEnableLmq(true);
+        messageStoreConfig.setEnableMultiDispatch(true);
+
+        BrokerConfig brokerConfig = new BrokerConfig();
+
+        DefaultMessageStore master = new DefaultMessageStore(
+            messageStoreConfig,
+            new BrokerStatsManager(brokerConfig.getBrokerClusterName(), brokerConfig.isEnableDetailStat()),
+            new MessageArrivingListener() {
+                @Override
+                public void arriving(String topic, int queueId, long logicOffset, long tagsCode,
+                    long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
+                }
+            }
+            , brokerConfig);
+
+        assertThat(master.load()).isTrue();
+
+        master.start();
+
+        return master;
+    }
+
    protected void putMsg(DefaultMessageStore master) throws Exception {
        long totalMsgs = 200;

@@ -158,6 +187,33 @@ public class ConsumeQueueTest {
        }
    }

+    protected void putMsgMultiQueue(DefaultMessageStore master) throws Exception {
+        for (long i = 0; i < 1; i++) {
+            master.putMessage(buildMessageMultiQueue());
+        }
+    }
+
+    private MessageExtBrokerInner buildMessageMultiQueue() {
+        MessageExtBrokerInner msg = new MessageExtBrokerInner();
+        msg.setTopic(topic);
+        msg.setTags("TAG1");
+        msg.setKeys("Hello");
+        msg.setBody(msgBody);
+        msg.setKeys(String.valueOf(System.currentTimeMillis()));
+        msg.setQueueId(queueId);
+        msg.setSysFlag(0);
+        msg.setBornTimestamp(System.currentTimeMillis());
+        msg.setStoreHost(StoreHost);
+        msg.setBornHost(BornHost);
+        for (int i = 0; i < 1; i++) {
+            msg.putUserProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH, "%LMQ%123,%LMQ%456");
+            msg.putUserProperty(String.valueOf(i), "imagoodperson" + i);
+        }
+        msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
+
+        return msg;
+    }
+
    protected void deleteDirectory(String rootPath) {
        File file = new File(rootPath);
        deleteFile(file);
@@ -217,6 +273,89 @@ public class ConsumeQueueTest {

    }

+    @Test
+    public void testPutMessagePositionInfoWrapper_MultiQueue() throws Exception {
+        DefaultMessageStore messageStore = null;
+        try {
+            messageStore = genForMultiQueue();
+
+
+            int totalMessages = 10;
+
+            for (int i = 0; i < totalMessages; i++) {
+                putMsgMultiQueue(messageStore);
+            }
+            Thread.sleep(5);
+
+            ConsumeQueue cq = messageStore.getConsumeQueueTable().get(topic).get(queueId);
+            Method method = cq.getClass().getDeclaredMethod("putMessagePositionInfoWrapper", DispatchRequest.class, boolean.class);
+
+            assertThat(method).isNotNull();
+
+            method.setAccessible(true);
+
+            SelectMappedBufferResult result = messageStore.getCommitLog().getData(0);
+            assertThat(result != null).isTrue();
+
+            DispatchRequest dispatchRequest = messageStore.getCommitLog().checkMessageAndReturnSize(result.getByteBuffer(), false, false);
+
+            assertThat(cq).isNotNull();
+
+            Object dispatchResult = method.invoke(cq,  dispatchRequest, true);
+
+            ConsumeQueue lmqCq1 = messageStore.getConsumeQueueTable().get("%LMQ%123").get(0);
+
+            ConsumeQueue lmqCq2 = messageStore.getConsumeQueueTable().get("%LMQ%456").get(0);
+
+            assertThat(lmqCq1).isNotNull();
+
+            assertThat(lmqCq2).isNotNull();
+
+        } finally {
+            if (messageStore != null) {
+                messageStore.shutdown();
+                messageStore.destroy();
+            }
+            deleteDirectory(storePath);
+        }
+
+    }
+
+    @Test
+    public void testPutMessagePositionInfoMultiQueue() throws Exception {
+        DefaultMessageStore messageStore = null;
+        try {
+
+            messageStore = genForMultiQueue();
+
+            int totalMessages = 10;
+
+            for (int i = 0; i < totalMessages; i++) {
+                putMsgMultiQueue(messageStore);
+            }
+            Thread.sleep(5);
+
+            ConsumeQueue cq = messageStore.getConsumeQueueTable().get(topic).get(queueId);
+
+            ConsumeQueue lmqCq1 = messageStore.getConsumeQueueTable().get("%LMQ%123").get(0);
+
+            ConsumeQueue lmqCq2 = messageStore.getConsumeQueueTable().get("%LMQ%456").get(0);
+
+            assertThat(cq).isNotNull();
+
+            assertThat(lmqCq1).isNotNull();
+
+            assertThat(lmqCq2).isNotNull();
+
+        } finally {
+            if (messageStore != null) {
+                messageStore.shutdown();
+                messageStore.destroy();
+            }
+            deleteDirectory(storePath);
+        }
+    }
+
    @Test
    public void testConsumeQueueWithExtendData() {
        DefaultMessageStore master = null;
@@ -34,6 +34,7 @@ import java.util.Random;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.store.config.FlushDiskType;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
@@ -641,6 +642,21 @@ public class DefaultMessageStoreTest {
        fileChannel.close();
    }

+    @Test
+    public void testCleanUnusedLmqTopic() throws Exception {
+        String lmqTopic = "%LMQ%123";
+
+        MessageExtBrokerInner messageExtBrokerInner = buildMessage();
+        messageExtBrokerInner.setTopic("test");
+        messageExtBrokerInner.setQueueId(0);
+        messageExtBrokerInner.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_DISPATCH, lmqTopic);
+        messageStore.putMessage(messageExtBrokerInner);
+
+        Thread.sleep(3000);
+        messageStore.cleanUnusedLmqTopic(lmqTopic);
+
+    }
+
    private class MyMessageArrivingListener implements MessageArrivingListener {
        @Override
        public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime,
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.store;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.nio.charset.Charset;
+
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.store.config.MessageStoreConfig;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class MultiDispatchTest {
+
+    private CommitLog commitLog;
+    private MultiDispatch multiDispatch;
+
+    @Before
+    public void init() throws Exception {
+        MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+        messageStoreConfig.setMappedFileSizeCommitLog(1024 * 8);
+        messageStoreConfig.setMappedFileSizeConsumeQueue(1024 * 4);
+        messageStoreConfig.setMaxHashSlotNum(100);
+        messageStoreConfig.setMaxIndexNum(100 * 10);
+        messageStoreConfig.setStorePathRootDir(System.getProperty("user.home") + File.separator + "unitteststore1");
+        messageStoreConfig.setStorePathCommitLog(
+            System.getProperty("user.home") + File.separator + "unitteststore1" + File.separator + "commitlog");
+
+        messageStoreConfig.setEnableLmq(true);
+        messageStoreConfig.setEnableMultiDispatch(true);
+        //too much reference
+        DefaultMessageStore messageStore = new DefaultMessageStore(messageStoreConfig, null, null, null);
+        this.commitLog = new CommitLog(messageStore);
+        this.multiDispatch = new MultiDispatch(messageStore, commitLog);
+    }
+
+    @After
+    public void destroy() {
+        UtilAll.deleteFile(new File(System.getProperty("user.home") + File.separator + "unitteststore1"));
+    }
+
+    @Test
+    public void queueKey() {
+        MessageExtBrokerInner messageExtBrokerInner = mock(MessageExtBrokerInner.class);
+        when(messageExtBrokerInner.getQueueId()).thenReturn(2);
+        String ret = multiDispatch.queueKey("%LMQ%lmq123", messageExtBrokerInner);
+        assertEquals(ret, "%LMQ%lmq123-0");
+    }
+
+    @Test
+    public void wrapMultiDispatch() {
+        MessageExtBrokerInner messageExtBrokerInner = mock(MessageExtBrokerInner.class);
+        when(messageExtBrokerInner.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH)).thenReturn(
+            "%LMQ%123,%LMQ%456");
+        when(messageExtBrokerInner.getTopic()).thenReturn("test");
+        when(messageExtBrokerInner.getBody()).thenReturn("aaa".getBytes(Charset.forName("UTF-8")));
+        when(messageExtBrokerInner.getBornHost()).thenReturn(new InetSocketAddress("127.0.0.1", 54270));
+        when(messageExtBrokerInner.getStoreHost()).thenReturn(new InetSocketAddress("127.0.0.1", 10911));
+        multiDispatch.wrapMultiDispatch(messageExtBrokerInner);
+        assertTrue(commitLog.getLmqTopicQueueTable().size() == 2);
+        assertTrue(commitLog.getLmqTopicQueueTable().get("%LMQ%123-0") == 0L);
+        assertTrue(commitLog.getLmqTopicQueueTable().get("%LMQ%456-0") == 0L);
+    }
+
+    @Test
+    public void updateMultiQueueOffset() {
+        MessageExtBrokerInner messageExtBrokerInner = mock(MessageExtBrokerInner.class);
+        when(messageExtBrokerInner.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH)).thenReturn("%LMQ%123,%LMQ%456");
+        when(messageExtBrokerInner.getProperty(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET)).thenReturn("0,1");
+        multiDispatch.updateMultiQueueOffset(messageExtBrokerInner);
+        assertTrue(commitLog.getLmqTopicQueueTable().size() == 2);
+        assertTrue(commitLog.getLmqTopicQueueTable().get("%LMQ%123-0") == 1L);
+        assertTrue(commitLog.getLmqTopicQueueTable().get("%LMQ%456-0") == 2L);
+    }
+}
@@ -45,6 +45,27 @@
            <artifactId>truth</artifactId>
            <version>0.30</version>
        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <version>2.28.2</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-client</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-tools</artifactId>
+        </dependency>
+
    </dependencies>

    <build>
@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.test.lmq.benchmark;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.PullCallback;
+import org.apache.rocketmq.client.consumer.PullResult;
+import org.apache.rocketmq.client.consumer.PullStatus;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.ThreadFactoryImpl;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.header.QueryConsumerOffsetRequestHeader;
+import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
+import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.test.util.StatUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+public class BenchLmqStore {
+    private static Logger logger = LoggerFactory.getLogger(BenchLmqStore.class);
+    private static String namesrv = System.getProperty("namesrv", "127.0.0.1:9876");
+    private static String lmqTopic = System.getProperty("lmqTopic", "lmqTestTopic");
+    private static boolean enableSub = Boolean.parseBoolean(System.getProperty("enableSub", "true"));
+    private static String queuePrefix = System.getProperty("queuePrefix", "lmqTest");
+    private static int tps = Integer.parseInt(System.getProperty("tps", "1"));
+    private static int lmqNum = Integer.parseInt(System.getProperty("lmqNum", "1"));
+    private static int sendThreadNum = Integer.parseInt(System.getProperty("sendThreadNum", "64"));
+    private static int consumerThreadNum = Integer.parseInt(System.getProperty("consumerThreadNum", "64"));
+    private static String brokerName = System.getProperty("brokerName", "broker-a");
+    private static int size = Integer.parseInt(System.getProperty("size", "128"));
+    private static int suspendTime = Integer.parseInt(System.getProperty("suspendTime", "2000"));
+    private static final boolean RETRY_NO_MATCHED_MSG = Boolean.parseBoolean(System.getProperty("retry_no_matched_msg", "false"));
+    private static boolean benchOffset = Boolean.parseBoolean(System.getProperty("benchOffset", "false"));
+    private static int benchOffsetNum = Integer.parseInt(System.getProperty("benchOffsetNum", "1"));
+    private static Map<MessageQueue, Long> offsetMap = new ConcurrentHashMap<>(256);
+    private static Map<MessageQueue, Boolean> pullStatus = new ConcurrentHashMap<>(256);
+    private static Map<Integer, Map<MessageQueue, Long>> pullEvent = new ConcurrentHashMap<>(256);
+    public static DefaultMQProducer defaultMQProducer;
+    private static int pullConsumerNum = Integer.parseInt(System.getProperty("pullConsumerNum", "8"));
+    public static DefaultMQPullConsumer[] defaultMQPullConsumers = new DefaultMQPullConsumer[pullConsumerNum];
+    private static AtomicLong rid = new AtomicLong();
+    private static final String LMQ_PREFIX = "%LMQ%";
+
+    public static void main(String[] args) throws InterruptedException, MQClientException, MQBrokerException,
+        RemotingException {
+        defaultMQProducer = new DefaultMQProducer();
+        defaultMQProducer.setProducerGroup("PID_LMQ_TEST");
+        defaultMQProducer.setVipChannelEnabled(false);
+        defaultMQProducer.setNamesrvAddr(namesrv);
+        defaultMQProducer.start();
+        //defaultMQProducer.createTopic(lmqTopic, lmqTopic, 8);
+        for (int i = 0; i < pullConsumerNum; i++) {
+            DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer();
+            defaultMQPullConsumers[i] = defaultMQPullConsumer;
+            defaultMQPullConsumer.setNamesrvAddr(namesrv);
+            defaultMQPullConsumer.setVipChannelEnabled(false);
+            defaultMQPullConsumer.setConsumerGroup("CID_RMQ_SYS_LMQ_TEST_" + i);
+            defaultMQPullConsumer.setInstanceName("CID_RMQ_SYS_LMQ_TEST_" + i);
+            defaultMQPullConsumer.setRegisterTopics(new HashSet<>(Arrays.asList(lmqTopic)));
+            defaultMQPullConsumer.setBrokerSuspendMaxTimeMillis(suspendTime);
+            defaultMQPullConsumer.setConsumerTimeoutMillisWhenSuspend(suspendTime + 1000);
+            defaultMQPullConsumer.start();
+        }
+        Thread.sleep(3000L);
+        if (benchOffset) {
+            doBenchOffset();
+            return;
+        }
+        ScheduledThreadPoolExecutor consumerPool = new ScheduledThreadPoolExecutor(consumerThreadNum, new ThreadFactoryImpl("test"));
+        for (int i = 0; i < consumerThreadNum; i++) {
+            final int idx = i;
+            consumerPool.scheduleWithFixedDelay(() -> {
+                try {
+                    Map<MessageQueue, Long> map = pullEvent.get(idx);
+                    if (map == null) {
+                        return;
+                    }
+                    for (Map.Entry<MessageQueue, Long> entry : map.entrySet()) {
+                        try {
+                            Boolean status = pullStatus.get(entry.getKey());
+                            if (Boolean.TRUE.equals(status)) {
+                                continue;
+                            }
+                            doPull(map, entry.getKey(), entry.getValue());
+                        } catch (Exception e) {
+                            logger.error("pull broker msg error", e);
+                        }
+                    }
+                } catch (Exception e) {
+                    logger.error("exec doPull task error", e);
+                }
+            }, 1, 1, TimeUnit.MILLISECONDS);
+        }
+        // init queue sub
+        if (enableSub && lmqNum > 0 && StringUtils.isNotBlank(brokerName)) {
+            for (int i = 0; i < lmqNum; i++) {
+                long idx = rid.incrementAndGet();
+                String queue = LMQ_PREFIX + queuePrefix + idx % lmqNum;
+                MessageQueue mq = new MessageQueue(queue, brokerName, 0);
+                int queueHash = Math.abs(queue.hashCode()) % consumerThreadNum;
+                pullEvent.putIfAbsent(queueHash, new ConcurrentHashMap<>());
+                pullEvent.get(queueHash).put(mq, idx);
+            }
+        }
+        Thread.sleep(5000L);
+        doSend();
+    }
+    public static void doSend() {
+        StringBuilder sb = new StringBuilder();
+        for (int j = 0; j < size; j += 10) {
+            sb.append("hello baby");
+        }
+        byte[] body = sb.toString().getBytes();
+        String pubKey = "pub";
+        ExecutorService sendPool = Executors.newFixedThreadPool(sendThreadNum);
+        for (int i = 0; i < sendThreadNum; i++) {
+            sendPool.execute(() -> {
+                while (true) {
+                    if (StatUtil.isOverFlow(pubKey, tps)) {
+                        try {
+                            Thread.sleep(100L);
+                        } catch (InterruptedException e) {
+                            e.printStackTrace();
+                        }
+                    }
+                    long start = System.currentTimeMillis();
+                    try {
+                        long idx = rid.incrementAndGet();
+                        Message message = new Message(lmqTopic, body);
+                        String queue = lmqTopic;
+                        if (lmqNum > 0) {
+                            queue = LMQ_PREFIX + queuePrefix + idx % lmqNum;
+                            message.putUserProperty("INNER_MULTI_DISPATCH", queue);
+                        }
+                        SendResult sendResult = defaultMQProducer.send(message);
+                        StatUtil.addInvoke(pubKey, System.currentTimeMillis() - start);
+                        if (StatUtil.nowTps(pubKey) < 10) {
+                            logger.warn("pub: {} ", sendResult.getMsgId());
+                        }
+                        if (enableSub) {
+                            MessageQueue mq = new MessageQueue(queue, sendResult.getMessageQueue().getBrokerName(),
+                                    lmqNum > 0 ? 0 : sendResult.getMessageQueue().getQueueId());
+                            int queueHash = Math.abs(queue.hashCode()) % consumerThreadNum;
+                            pullEvent.putIfAbsent(queueHash, new ConcurrentHashMap<>());
+                            pullEvent.get(queueHash).put(mq, idx);
+                        }
+                    } catch (Exception e) {
+                        logger.error("", e);
+                        StatUtil.addInvoke(pubKey, System.currentTimeMillis() - start, false);
+                    }
+                }
+            });
+        }
+    }
+    public static void doPull(Map<MessageQueue, Long> eventMap, MessageQueue mq, Long eventId) throws RemotingException, InterruptedException, MQClientException {
+        if (!enableSub) {
+            eventMap.remove(mq, eventId);
+            pullStatus.remove(mq);
+            return;
+        }
+        DefaultMQPullConsumer defaultMQPullConsumer = defaultMQPullConsumers[(int) (eventId % pullConsumerNum)];
+        Long offset = offsetMap.get(mq);
+        if (offset == null) {
+            long start = System.currentTimeMillis();
+            offset = defaultMQPullConsumer.maxOffset(mq);
+            StatUtil.addInvoke("maxOffset", System.currentTimeMillis() - start);
+            offsetMap.put(mq, offset);
+        }
+        long start = System.currentTimeMillis();
+        if (null != pullStatus.putIfAbsent(mq, true)) {
+            return;
+        }
+        defaultMQPullConsumer.pullBlockIfNotFound(
+                mq, "*", offset, 32,
+                new PullCallback() {
+                    @Override
+                    public void onSuccess(PullResult pullResult) {
+                        StatUtil.addInvoke(pullResult.getPullStatus().name(), System.currentTimeMillis() - start);
+                        eventMap.remove(mq, eventId);
+                        pullStatus.remove(mq);
+                        offsetMap.put(mq, pullResult.getNextBeginOffset());
+                        StatUtil.addInvoke("doPull", System.currentTimeMillis() - start);
+                        if (PullStatus.NO_MATCHED_MSG.equals(pullResult.getPullStatus()) && RETRY_NO_MATCHED_MSG) {
+                            long idx = rid.incrementAndGet();
+                            eventMap.put(mq, idx);
+                        }
+                        List<MessageExt> list = pullResult.getMsgFoundList();
+                        if (list == null || list.isEmpty()) {
+                            StatUtil.addInvoke("NoMsg", System.currentTimeMillis() - start);
+                            return;
+                        }
+                        for (MessageExt messageExt : list) {
+                            StatUtil.addInvoke("sub", System.currentTimeMillis() - messageExt.getBornTimestamp());
+                            if (StatUtil.nowTps("sub") < 10) {
+                                logger.warn("sub: {}", messageExt.getMsgId());
+                            }
+                        }
+                    }
+                    @Override
+                    public void onException(Throwable e) {
+                        eventMap.remove(mq, eventId);
+                        pullStatus.remove(mq);
+                        logger.error("", e);
+                        StatUtil.addInvoke("doPull", System.currentTimeMillis() - start, false);
+                    }
+                });
+    }
+    public static void doBenchOffset() throws RemotingException, InterruptedException, MQClientException {
+        ExecutorService sendPool = Executors.newFixedThreadPool(sendThreadNum);
+        Map<String, Long> offsetMap = new ConcurrentHashMap<>();
+        String statKey = "benchOffset";
+        TopicRouteData topicRouteData = defaultMQPullConsumers[0].getDefaultMQPullConsumerImpl().
+                getRebalanceImpl().getMqClientFactory().getMQClientAPIImpl().
+                getTopicRouteInfoFromNameServer(lmqTopic, 3000);
+        HashMap<Long, String> brokerMap = topicRouteData.getBrokerDatas().get(0).getBrokerAddrs();
+        if (brokerMap == null || brokerMap.isEmpty()) {
+            return;
+        }
+        String brokerAddress = brokerMap.get(MixAll.MASTER_ID);
+        for (int i = 0; i < sendThreadNum; i++) {
+            final int flag = i;
+            sendPool.execute(new Runnable() {
+                @Override
+                public void run() {
+                    while (true) {
+                        try {
+                            if (StatUtil.isOverFlow(statKey, tps)) {
+                                Thread.sleep(100L);
+                            }
+                            long start = System.currentTimeMillis();
+                            DefaultMQPullConsumer defaultMQPullConsumer = defaultMQPullConsumers[(int) (rid.incrementAndGet() % pullConsumerNum)];
+                            long id = rid.incrementAndGet();
+                            String lmq = LMQ_PREFIX + queuePrefix + id % benchOffsetNum;
+                            String lmqCid = LMQ_PREFIX + "GID_LMQ@@c" + flag + "-" + id % benchOffsetNum;
+                            Long offset = offsetMap.get(lmq);
+                            if (offset == null) {
+                                offsetMap.put(lmq, 0L);
+                            }
+                            long newOffset1 = offsetMap.get(lmq) + 1;
+                            UpdateConsumerOffsetRequestHeader updateHeader = new UpdateConsumerOffsetRequestHeader();
+                            updateHeader.setTopic(lmq);
+                            updateHeader.setConsumerGroup(lmqCid);
+                            updateHeader.setQueueId(0);
+                            updateHeader.setCommitOffset(newOffset1);
+                            defaultMQPullConsumer
+                                    .getDefaultMQPullConsumerImpl()
+                                    .getRebalanceImpl()
+                                    .getMqClientFactory()
+                                    .getMQClientAPIImpl().updateConsumerOffset(brokerAddress, updateHeader, 1000);
+                            QueryConsumerOffsetRequestHeader queryHeader = new QueryConsumerOffsetRequestHeader();
+                            queryHeader.setTopic(lmq);
+                            queryHeader.setConsumerGroup(lmqCid);
+                            queryHeader.setQueueId(0);
+                            long newOffset2 = defaultMQPullConsumer
+                                    .getDefaultMQPullConsumerImpl()
+                                    .getRebalanceImpl()
+                                    .getMqClientFactory()
+                                    .getMQClientAPIImpl()
+                                    .queryConsumerOffset(brokerAddress, queryHeader, 1000);
+                            offsetMap.put(lmq, newOffset2);
+                            if (newOffset1 != newOffset2) {
+                                StatUtil.addInvoke("ErrorOffset", 1);
+                            }
+                            StatUtil.addInvoke(statKey, System.currentTimeMillis() - start);
+                        } catch (Exception e) {
+                            logger.error("", e);
+                        }
+                    }
+                }
+            });
+        }
+    }
+}
@@ -0,0 +1,478 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.test.util;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Generated;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static java.math.BigDecimal.ROUND_HALF_UP;
+
+@Generated("StatUtil")
+public class StatUtil {
+    private static Logger sysLogger = LoggerFactory.getLogger(StatUtil.class);
+    private static Logger logger = LoggerFactory.getLogger("StatLogger");
+    private static final int MAX_KEY_NUM = Integer.parseInt(System.getProperty("stat.util.key.max.num", "10000"));
+    private static volatile ConcurrentMap<String, Invoke> invokeCache = new ConcurrentHashMap<>(64);
+    private static volatile ConcurrentMap<String, Map<Long, SecondInvoke>> secondInvokeCache = new ConcurrentHashMap<>(
+        64);
+
+    private static final int STAT_WINDOW_SECONDS = Integer.parseInt(System.getProperty("stat.win.seconds", "60"));
+    private static final String SPLITTER = "|";
+    private static ScheduledExecutorService daemon = Executors.newSingleThreadScheduledExecutor();
+
+    static class Invoke {
+        AtomicLong totalPv = new AtomicLong();
+        AtomicLong failPv = new AtomicLong();
+        AtomicLong sumRt = new AtomicLong();
+        AtomicLong maxRt = new AtomicLong();
+        AtomicLong minRt = new AtomicLong();
+        AtomicInteger topSecondPv = new AtomicInteger();
+        AtomicInteger secondPv = new AtomicInteger();
+        AtomicLong second = new AtomicLong(System.currentTimeMillis() / 1000L);
+    }
+
+    static class SecondInvoke implements Comparable<SecondInvoke> {
+        AtomicLong total = new AtomicLong();
+        AtomicLong fail = new AtomicLong();
+        AtomicLong sumRt = new AtomicLong();
+        AtomicLong maxRt = new AtomicLong();
+        AtomicLong minRt = new AtomicLong();
+        Long second = nowSecond();
+
+        @Override
+        public int compareTo(SecondInvoke o) {
+            return o.second.compareTo(second);
+        }
+    }
+
+    static {
+        daemon.scheduleAtFixedRate(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    printInvokeStat();
+                    printSecondInvokeStat();
+                } catch (Exception e) {
+                    logger.error("", e);
+                }
+            }
+        }, STAT_WINDOW_SECONDS, STAT_WINDOW_SECONDS, TimeUnit.SECONDS);
+    }
+
+    private static void printInvokeStat() {
+        Map<String, Invoke> tmp = invokeCache;
+        invokeCache = new ConcurrentHashMap<>(64);
+
+        sysLogger.warn("printInvokeStat key count:{}", tmp.size());
+        for (Map.Entry<String, Invoke> entry : tmp.entrySet()) {
+            String key = entry.getKey();
+            Invoke invoke = entry.getValue();
+            logger.warn("{}",
+                buildLog(key, invoke.topSecondPv.get(), invoke.totalPv.get(), invoke.failPv.get(), invoke.minRt.get(),
+                    invoke.maxRt.get(), invoke.sumRt.get()));
+        }
+    }
+
+    private static void printSecondInvokeStat() {
+        sysLogger.warn("printSecondInvokeStat key count:{}", secondInvokeCache.size());
+        for (Map.Entry<String, Map<Long, SecondInvoke>> entry : secondInvokeCache.entrySet()) {
+            String key = entry.getKey();
+            Map<Long, SecondInvoke> secondInvokeMap = entry.getValue();
+            long totalPv = 0L;
+            long failPv = 0L;
+            long topSecondPv = 0L;
+            long sumRt = 0L;
+            long maxRt = 0L;
+            long minRt = 0L;
+
+            for (Map.Entry<Long, SecondInvoke> invokeEntry : secondInvokeMap.entrySet()) {
+                long second = invokeEntry.getKey();
+                SecondInvoke secondInvoke = invokeEntry.getValue();
+                if (nowSecond() - second >= STAT_WINDOW_SECONDS) {
+                    secondInvokeMap.remove(second);
+                    continue;
+                }
+                long secondPv = secondInvoke.total.get();
+                totalPv += secondPv;
+                failPv += secondInvoke.fail.get();
+                sumRt += secondInvoke.sumRt.get();
+                if (maxRt < secondInvoke.maxRt.get()) {
+                    maxRt = secondInvoke.maxRt.get();
+                }
+                if (minRt > secondInvoke.minRt.get()) {
+                    minRt = secondInvoke.minRt.get();
+                }
+                if (topSecondPv < secondPv) {
+                    topSecondPv = secondPv;
+                }
+            }
+            if (secondInvokeMap.isEmpty()) {
+                secondInvokeCache.remove(key);
+                continue;
+            }
+            logger.warn("{}", buildLog(key, topSecondPv, totalPv, failPv, minRt, maxRt, sumRt));
+        }
+    }
+
+    private static String buildLog(String key, long topSecondPv, long totalPv, long failPv, long minRt, long maxRt,
+        long sumRt) {
+        StringBuilder sb = new StringBuilder();
+        sb.append(SPLITTER);
+        sb.append(key);
+        sb.append(SPLITTER);
+        sb.append(topSecondPv);
+        sb.append(SPLITTER);
+        int tps = new BigDecimal(totalPv).divide(new BigDecimal(STAT_WINDOW_SECONDS),
+            ROUND_HALF_UP).intValue();
+        sb.append(tps);
+        sb.append(SPLITTER);
+        sb.append(totalPv);
+        sb.append(SPLITTER);
+        sb.append(failPv);
+        sb.append(SPLITTER);
+        sb.append(minRt);
+        sb.append(SPLITTER);
+        long avg = new BigDecimal(sumRt).divide(new BigDecimal(totalPv),
+            ROUND_HALF_UP).longValue();
+        sb.append(avg);
+        sb.append(SPLITTER);
+        sb.append(maxRt);
+        return sb.toString();
+    }
+
+    public static String buildKey(String... keys) {
+        if (keys == null || keys.length <= 0) {
+            return null;
+        }
+        StringBuilder sb = new StringBuilder();
+        for (String key : keys) {
+            sb.append(key);
+            sb.append(",");
+        }
+        sb.deleteCharAt(sb.length() - 1);
+        return sb.toString();
+    }
+
+    public static void addInvoke(String key, long rt) {
+        addInvoke(key, rt, true);
+    }
+
+    private static Invoke getAndSetInvoke(String key) {
+        Invoke invoke = invokeCache.get(key);
+        if (invoke == null) {
+            invokeCache.putIfAbsent(key, new Invoke());
+        }
+        return invokeCache.get(key);
+    }
+
+    public static void addInvoke(String key, int num, long rt, boolean success) {
+        if (invokeCache.size() > MAX_KEY_NUM || secondInvokeCache.size() > MAX_KEY_NUM) {
+            return;
+        }
+        Invoke invoke = getAndSetInvoke(key);
+        if (invoke == null) {
+            return;
+        }
+
+        invoke.totalPv.getAndAdd(num);
+        if (!success) {
+            invoke.failPv.getAndAdd(num);
+        }
+        long now = nowSecond();
+        AtomicLong oldSecond = invoke.second;
+        if (oldSecond.get() == now) {
+            invoke.secondPv.getAndAdd(num);
+        } else {
+            if (oldSecond.compareAndSet(oldSecond.get(), now)) {
+                if (invoke.secondPv.get() > invoke.topSecondPv.get()) {
+                    invoke.topSecondPv.set(invoke.secondPv.get());
+                }
+                invoke.secondPv.set(num);
+            } else {
+                invoke.secondPv.getAndAdd(num);
+            }
+        }
+
+        invoke.sumRt.addAndGet(rt);
+        if (invoke.maxRt.get() < rt) {
+            invoke.maxRt.set(rt);
+        }
+        if (invoke.minRt.get() > rt) {
+            invoke.minRt.set(rt);
+        }
+    }
+
+    public static void addInvoke(String key, long rt, boolean success) {
+        if (invokeCache.size() > MAX_KEY_NUM || secondInvokeCache.size() > MAX_KEY_NUM) {
+            return;
+        }
+        Invoke invoke = getAndSetInvoke(key);
+        if (invoke == null) {
+            return;
+        }
+
+        invoke.totalPv.getAndIncrement();
+        if (!success) {
+            invoke.failPv.getAndIncrement();
+        }
+        long now = nowSecond();
+        AtomicLong oldSecond = invoke.second;
+        if (oldSecond.get() == now) {
+            invoke.secondPv.getAndIncrement();
+        } else {
+            if (oldSecond.compareAndSet(oldSecond.get(), now)) {
+                if (invoke.secondPv.get() > invoke.topSecondPv.get()) {
+                    invoke.topSecondPv.set(invoke.secondPv.get());
+                }
+                invoke.secondPv.set(1);
+            } else {
+                invoke.secondPv.getAndIncrement();
+            }
+        }
+
+        invoke.sumRt.addAndGet(rt);
+        if (invoke.maxRt.get() < rt) {
+            invoke.maxRt.set(rt);
+        }
+        if (invoke.minRt.get() > rt) {
+            invoke.minRt.set(rt);
+        }
+    }
+
+    public static SecondInvoke getAndSetSecondInvoke(String key) {
+        if (!secondInvokeCache.containsKey(key)) {
+            secondInvokeCache.putIfAbsent(key, new ConcurrentHashMap<>(STAT_WINDOW_SECONDS));
+        }
+        Map<Long, SecondInvoke> secondInvokeMap = secondInvokeCache.get(key);
+        if (secondInvokeMap == null) {
+            return null;
+        }
+        long second = nowSecond();
+        if (!secondInvokeMap.containsKey(second)) {
+            secondInvokeMap.putIfAbsent(second, new SecondInvoke());
+        }
+        return secondInvokeMap.get(second);
+    }
+
+    public static void addSecondInvoke(String key, long rt) {
+        addSecondInvoke(key, rt, true);
+    }
+
+    public static void addSecondInvoke(String key, long rt, boolean success) {
+        if (invokeCache.size() > MAX_KEY_NUM || secondInvokeCache.size() > MAX_KEY_NUM) {
+            return;
+        }
+        SecondInvoke secondInvoke = getAndSetSecondInvoke(key);
+        if (secondInvoke == null) {
+            return;
+        }
+        secondInvoke.total.addAndGet(1);
+        if (!success) {
+            secondInvoke.fail.addAndGet(1);
+        }
+        secondInvoke.sumRt.addAndGet(rt);
+        if (secondInvoke.maxRt.get() < rt) {
+            secondInvoke.maxRt.set(rt);
+        }
+        if (secondInvoke.minRt.get() > rt) {
+            secondInvoke.minRt.set(rt);
+        }
+    }
+
+    public static void addPv(String key, long totalPv) {
+        addPv(key, totalPv, true);
+    }
+
+    public static void addPv(String key, long totalPv, boolean success) {
+        if (invokeCache.size() > MAX_KEY_NUM || secondInvokeCache.size() > MAX_KEY_NUM) {
+            return;
+        }
+        if (totalPv <= 0) {
+            return;
+        }
+        Invoke invoke = getAndSetInvoke(key);
+        if (invoke == null) {
+            return;
+        }
+        invoke.totalPv.addAndGet(totalPv);
+        if (!success) {
+            invoke.failPv.addAndGet(totalPv);
+        }
+        long now = nowSecond();
+        AtomicLong oldSecond = invoke.second;
+        if (oldSecond.get() == now) {
+            invoke.secondPv.addAndGet((int)totalPv);
+        } else {
+            if (oldSecond.compareAndSet(oldSecond.get(), now)) {
+                if (invoke.secondPv.get() > invoke.topSecondPv.get()) {
+                    invoke.topSecondPv.set(invoke.secondPv.get());
+                }
+                invoke.secondPv.set((int)totalPv);
+            } else {
+                invoke.secondPv.addAndGet((int)totalPv);
+            }
+        }
+    }
+
+    public static void addSecondPv(String key, long totalPv) {
+        addSecondPv(key, totalPv, true);
+    }
+
+    public static void addSecondPv(String key, long totalPv, boolean success) {
+        if (invokeCache.size() > MAX_KEY_NUM || secondInvokeCache.size() > MAX_KEY_NUM) {
+            return;
+        }
+        if (totalPv <= 0) {
+            return;
+        }
+        SecondInvoke secondInvoke = getAndSetSecondInvoke(key);
+        if (secondInvoke == null) {
+            return;
+        }
+        secondInvoke.total.addAndGet(totalPv);
+        if (!success) {
+            secondInvoke.fail.addAndGet(totalPv);
+        }
+    }
+
+    public static boolean isOverFlow(String key, int tps) {
+        return nowTps(key) >= tps;
+    }
+
+    public static int nowTps(String key) {
+        Map<Long, SecondInvoke> secondInvokeMap = secondInvokeCache.get(key);
+        if (secondInvokeMap != null) {
+            SecondInvoke secondInvoke = secondInvokeMap.get(nowSecond());
+            if (secondInvoke != null) {
+                return (int)secondInvoke.total.get();
+            }
+        }
+        Invoke invoke = invokeCache.get(key);
+        if (invoke == null) {
+            return 0;
+        }
+        AtomicLong oldSecond = invoke.second;
+        if (oldSecond.get() == nowSecond()) {
+            return invoke.secondPv.get();
+        }
+        return 0;
+    }
+
+    public static int totalPvInWindow(String key, int windowSeconds) {
+        List<SecondInvoke> list = secondInvokeList(key, windowSeconds);
+        long totalPv = 0;
+        for (int i = 0; i < windowSeconds && i < list.size(); i++) {
+            totalPv += list.get(i).total.get();
+        }
+        return (int)totalPv;
+    }
+
+    public static int failPvInWindow(String key, int windowSeconds) {
+        List<SecondInvoke> list = secondInvokeList(key, windowSeconds);
+        long failPv = 0;
+        for (int i = 0; i < windowSeconds && i < list.size(); i++) {
+            failPv += list.get(i).fail.get();
+        }
+        return (int)failPv;
+    }
+
+    public static int topTpsInWindow(String key, int windowSeconds) {
+        List<SecondInvoke> list = secondInvokeList(key, windowSeconds);
+        long topTps = 0;
+        for (int i = 0; i < windowSeconds && i < list.size(); i++) {
+            long secondPv = list.get(i).total.get();
+            if (topTps < secondPv) {
+                topTps = secondPv;
+            }
+        }
+        return (int)topTps;
+    }
+
+    public static int avgRtInWindow(String key, int windowSeconds) {
+        List<SecondInvoke> list = secondInvokeList(key, windowSeconds);
+        long sumRt = 0;
+        long totalPv = 0;
+        for (int i = 0; i < windowSeconds && i < list.size(); i++) {
+            sumRt += list.get(i).sumRt.get();
+            totalPv += list.get(i).total.get();
+        }
+        if (totalPv <= 0) {
+            return 0;
+        }
+        long avg = new BigDecimal(sumRt).divide(new BigDecimal(totalPv),
+            ROUND_HALF_UP).longValue();
+        return (int)avg;
+    }
+
+    public static int maxRtInWindow(String key, int windowSeconds) {
+        List<SecondInvoke> list = secondInvokeList(key, windowSeconds);
+        long maxRt = 0;
+        long totalPv = 0;
+        for (int i = 0; i < windowSeconds && i < list.size(); i++) {
+            if (maxRt < list.get(i).maxRt.get()) {
+                maxRt = list.get(i).maxRt.get();
+            }
+        }
+        return (int)maxRt;
+    }
+
+    public static int minRtInWindow(String key, int windowSeconds) {
+        List<SecondInvoke> list = secondInvokeList(key, windowSeconds);
+        long minRt = 0;
+        long totalPv = 0;
+        for (int i = 0; i < windowSeconds && i < list.size(); i++) {
+            if (minRt < list.get(i).minRt.get()) {
+                minRt = list.get(i).minRt.get();
+            }
+        }
+        return (int)minRt;
+    }
+
+    private static List<SecondInvoke> secondInvokeList(String key, int windowSeconds) {
+        if (windowSeconds > STAT_WINDOW_SECONDS || windowSeconds <= 0) {
+            throw new IllegalArgumentException("windowSeconds Must Not be great than " + STAT_WINDOW_SECONDS);
+        }
+        Map<Long, SecondInvoke> secondInvokeMap = secondInvokeCache.get(key);
+        if (secondInvokeMap == null || secondInvokeMap.isEmpty()) {
+            return new ArrayList<>();
+        }
+        List<SecondInvoke> list = new ArrayList<>();
+        list.addAll(secondInvokeMap.values());
+        Collections.sort(list);
+        return list;
+    }
+
+    private static long nowSecond() {
+        return System.currentTimeMillis() / 1000L;
+    }
+
+}
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.test.lmq;
+
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.PullCallback;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.MQClientAPIImpl;
+import org.apache.rocketmq.client.impl.consumer.DefaultMQPullConsumerImpl;
+import org.apache.rocketmq.client.impl.consumer.RebalanceImpl;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.header.QueryConsumerOffsetRequestHeader;
+import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.test.lmq.benchmark.BenchLmqStore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestBenchLmqStore {
+    @Test
+    public void test() throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
+        System.setProperty("sendThreadNum", "1");
+        System.setProperty("pullConsumerNum", "1");
+        System.setProperty("consumerThreadNum", "1");
+        BenchLmqStore.defaultMQProducer = mock(DefaultMQProducer.class);
+        SendResult sendResult = new SendResult();
+        when(BenchLmqStore.defaultMQProducer.send(any(Message.class))).thenReturn(sendResult);
+        BenchLmqStore.doSend();
+        Thread.sleep(100L);
+        //verify(BenchLmqStore.defaultMQProducer, atLeastOnce()).send(any(Message.class));
+        BenchLmqStore.defaultMQPullConsumers = new DefaultMQPullConsumer[1];
+        BenchLmqStore.defaultMQPullConsumers[0] = mock(DefaultMQPullConsumer.class);
+        BenchLmqStore.doPull(new ConcurrentHashMap<>(), new MessageQueue(), 1L);
+        verify(BenchLmqStore.defaultMQPullConsumers[0], atLeastOnce()).pullBlockIfNotFound(any(MessageQueue.class), anyString(), anyLong(), anyInt(), any(
+            PullCallback.class));
+    }
+    @Test
+    public void testOffset() throws RemotingException, InterruptedException, MQClientException, MQBrokerException, IllegalAccessException {
+        System.setProperty("sendThreadNum", "1");
+        DefaultMQPullConsumer defaultMQPullConsumer = mock(DefaultMQPullConsumer.class);
+        BenchLmqStore.defaultMQPullConsumers = new DefaultMQPullConsumer[1];
+        BenchLmqStore.defaultMQPullConsumers[0] = defaultMQPullConsumer;
+        DefaultMQPullConsumerImpl defaultMQPullConsumerImpl = mock(DefaultMQPullConsumerImpl.class);
+        when(defaultMQPullConsumer.getDefaultMQPullConsumerImpl()).thenReturn(defaultMQPullConsumerImpl);
+        RebalanceImpl rebalanceImpl = mock(RebalanceImpl.class);
+        when(defaultMQPullConsumerImpl.getRebalanceImpl()).thenReturn(rebalanceImpl);
+        MQClientInstance mqClientInstance = mock(MQClientInstance.class);
+        when(rebalanceImpl.getMqClientFactory()).thenReturn(mqClientInstance);
+        MQClientAPIImpl mqClientAPI = mock(MQClientAPIImpl.class);
+        when(mqClientInstance.getMQClientAPIImpl()).thenReturn(mqClientAPI);
+        TopicRouteData topicRouteData = new TopicRouteData();
+        HashMap<Long, String> brokerAddrs = new HashMap<>();
+        brokerAddrs.put(MixAll.MASTER_ID, "test");
+        List<BrokerData> brokerData = Arrays.asList(new BrokerData("test", "test", brokerAddrs));
+        topicRouteData.setBrokerDatas(brokerData);
+        FieldUtils.writeStaticField(BenchLmqStore.class, "lmqTopic", "test", true);
+        when(mqClientAPI.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(topicRouteData);
+        BenchLmqStore.doBenchOffset();
+        Thread.sleep(100L);
+        verify(mqClientAPI, atLeastOnce()).queryConsumerOffset(anyString(), any(QueryConsumerOffsetRequestHeader.class), anyLong());
+        verify(mqClientAPI, atLeastOnce()).updateConsumerOffset(anyString(), any(UpdateConsumerOffsetRequestHeader.class), anyLong());
+    }
+}