lock.lock() before try

apache · Nov 26, 2023 · 1f41101 · 1f41101
1 parent e955e43
commit 1f41101
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 22 deletions.
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java
@@ -379,8 +379,8 @@ public void operationSucceed(RemotingCommand response) {
                                                     MessageDecoder.decodes(ByteBuffer.wrap(response.getBody()), true);
 
                                                 QueryResult qr = new QueryResult(responseHeader.getIndexLastUpdateTimestamp(), wrappers);
+                                                lock.writeLock().lock();
                                                 try {
-                                                    lock.writeLock().lock();
                                                     queryResultList.add(qr);
                                                 } finally {
                                                     lock.writeLock().unlock();

diff --git a/.../src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java b/.../src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java
@@ -487,8 +487,8 @@ public void run() {
                             long beginTimestamp = System.currentTimeMillis();
                             ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
                             boolean hasException = false;
+                            this.processQueue.getConsumeLock().lock();
                             try {
-                                this.processQueue.getConsumeLock().lock();
                                 if (this.processQueue.isDropped()) {
                                     log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}",
                                         this.messageQueue);

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ProcessQueue.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ProcessQueue.java
@@ -424,29 +424,33 @@ public void fillProcessQueueInfo(final ProcessQueueInfo info) {
         try {
             this.treeMapLock.readLock().lockInterruptibly();
 
-            if (!this.msgTreeMap.isEmpty()) {
-                info.setCachedMsgMinOffset(this.msgTreeMap.firstKey());
-                info.setCachedMsgMaxOffset(this.msgTreeMap.lastKey());
-                info.setCachedMsgCount(this.msgTreeMap.size());
-                info.setCachedMsgSizeInMiB((int) (this.msgSize.get() / (1024 * 1024)));
-            }
+            try {
+                if (!this.msgTreeMap.isEmpty()) {
+                    info.setCachedMsgMinOffset(this.msgTreeMap.firstKey());
+                    info.setCachedMsgMaxOffset(this.msgTreeMap.lastKey());
+                    info.setCachedMsgCount(this.msgTreeMap.size());
+                    info.setCachedMsgSizeInMiB((int) (this.msgSize.get() / (1024 * 1024)));
+                }
 
-            if (!this.consumingMsgOrderlyTreeMap.isEmpty()) {
-                info.setTransactionMsgMinOffset(this.consumingMsgOrderlyTreeMap.firstKey());
-                info.setTransactionMsgMaxOffset(this.consumingMsgOrderlyTreeMap.lastKey());
-                info.setTransactionMsgCount(this.consumingMsgOrderlyTreeMap.size());
-            }
+                if (!this.consumingMsgOrderlyTreeMap.isEmpty()) {
+                    info.setTransactionMsgMinOffset(this.consumingMsgOrderlyTreeMap.firstKey());
+                    info.setTransactionMsgMaxOffset(this.consumingMsgOrderlyTreeMap.lastKey());
+                    info.setTransactionMsgCount(this.consumingMsgOrderlyTreeMap.size());
+                }
 
-            info.setLocked(this.locked);
-            info.setTryUnlockTimes(this.tryUnlockTimes.get());
-            info.setLastLockTimestamp(this.lastLockTimestamp);
+                info.setLocked(this.locked);
+                info.setTryUnlockTimes(this.tryUnlockTimes.get());
+                info.setLastLockTimestamp(this.lastLockTimestamp);
 
-            info.setDroped(this.dropped);
-            info.setLastPullTimestamp(this.lastPullTimestamp);
-            info.setLastConsumeTimestamp(this.lastConsumeTimestamp);
-        } catch (Exception e) {
-        } finally {
-            this.treeMapLock.readLock().unlock();
+                info.setDroped(this.dropped);
+                info.setLastPullTimestamp(this.lastPullTimestamp);
+                info.setLastConsumeTimestamp(this.lastConsumeTimestamp);
+            } finally {
+                this.treeMapLock.readLock().unlock();
+            }
+
+        } catch (InterruptedException e) {
+            log.error("fillProcessQueueInfo lock error, ProcessQueueInfo={}", info, e);
         }
     }