opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-28 20:09:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: avoid memory overhead when there is large number of LMQ ConsumeQueue (#8956)

Browse Source
This commit is contained in:
Zhanhui Li
2024-11-20 14:55:30 +08:00
committed by GitHub
parent c13f051eb8
commit 8505482c0b
2 changed files with 12 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/queue/AbstractConsumeQueueStore.java
+5 -1
View File
@@ -39,7 +39,11 @@ public abstract class AbstractConsumeQueueStore implements ConsumeQueueStoreInte
public AbstractConsumeQueueStore(DefaultMessageStore messageStore) {
this.messageStore = messageStore;
this.messageStoreConfig = messageStore.getMessageStoreConfig();
this.consumeQueueTable = new ConcurrentHashMap<>(32);
if (messageStoreConfig.isEnableLmq()) {
this.consumeQueueTable = new ConcurrentHashMap<>(32_768);
} else {
this.consumeQueueTable = new ConcurrentHashMap<>(32);
}
}
@Override
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java
+7 -1
View File
@@ -480,7 +480,13 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
public ConsumeQueueInterface findOrCreateConsumeQueue(String topic, int queueId) {
ConcurrentMap<Integer, ConsumeQueueInterface> map = this.consumeQueueTable.get(topic);
if (null == map) {
ConcurrentMap<Integer, ConsumeQueueInterface> newMap = new ConcurrentHashMap<>(128);
ConcurrentMap<Integer, ConsumeQueueInterface> newMap;
if (MixAll.isLmq(topic)) {
// For LMQ, no need to over allocate internal hashtable
newMap = new ConcurrentHashMap<>(1, 1.0F);
} else {
newMap = new ConcurrentHashMap<>(8);
}
ConcurrentMap<Integer, ConsumeQueueInterface> oldMap = this.consumeQueueTable.putIfAbsent(topic, newMap);
if (oldMap != null) {
map = oldMap;