Add the configuration of topicQueueLock number to better support different scenarios (#7317)

Author: RongtongJin

store/src/main/java/org/apache/rocketmq/store/CommitLog.java

@@ -122,7 +122,7 @@ protected PutMessageThreadLocal initialValue() {
 
         this.flushDiskWatcher = new FlushDiskWatcher();
 
-        this.topicQueueLock = new TopicQueueLock();
+        this.topicQueueLock = new TopicQueueLock(messageStore.getMessageStoreConfig().getTopicQueueLockNum());
 
         this.commitLogSize = messageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
     }

store/src/main/java/org/apache/rocketmq/store/TopicQueueLock.java

@@ -34,6 +34,14 @@ public TopicQueueLock() {
         }
     }
 
+    public TopicQueueLock(int size) {
+        this.size = size;
+        this.lockList = new ArrayList<>(size);
+        for (int i = 0; i < this.size; i++) {
+            this.lockList.add(new ReentrantLock());
+        }
+    }
+
     public void lock(String topicQueueKey) {
         Lock lock = this.lockList.get((topicQueueKey.hashCode() & 0x7fffffff) % this.size);
         lock.lock();

store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java

@@ -401,6 +401,8 @@ public class MessageStoreConfig {
     private long memTableFlushInterval = 60 * 60 * 1000L;
     private boolean enableRocksDBLog = false;
 
+    private int topicQueueLockNum = 32;
+
     public boolean isDebugLockEnable() {
         return debugLockEnable;
     }
@@ -1751,4 +1753,12 @@ public boolean isEnableRocksDBLog() {
     public void setEnableRocksDBLog(boolean enableRocksDBLog) {
         this.enableRocksDBLog = enableRocksDBLog;
     }
+
+    public int getTopicQueueLockNum() {
+        return topicQueueLockNum;
+    }
+
+    public void setTopicQueueLockNum(int topicQueueLockNum) {
+        this.topicQueueLockNum = topicQueueLockNum;
+    }
 }