You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2019/02/05 20:51:38 UTC
[activemq-artemis] 01/03: ARTEMIS-2244 checkDepage method placed
outside CRITICAL_DELIVER avoid critical analyzer timeout
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 8799615a136946c39bb49bd4069cff2df0035997
Author: yb <17...@cn.suning.com>
AuthorDate: Thu Jan 31 17:20:06 2019 +0800
ARTEMIS-2244 checkDepage method placed outside CRITICAL_DELIVER avoid critical analyzer timeout
---
.../activemq/artemis/core/server/impl/QueueImpl.java | 18 ++++++++++++------
1 file changed, 12 insertions(+), 6 deletions(-)
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index 8209995..a61a3d6 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -2409,7 +2409,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
* This method will deliver as many messages as possible until all consumers are busy or there
* are no more matching or available messages.
*/
- private void deliver() {
+ private boolean deliver() {
if (logger.isDebugEnabled()) {
logger.debug(this + " doing deliver. messageReferences=" + messageReferences.size());
}
@@ -2430,7 +2430,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
deliverAsync();
- return;
+ return false;
}
if (System.currentTimeMillis() > timeout) {
@@ -2440,7 +2440,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
deliverAsync();
- return;
+ return false;
}
MessageReference ref;
@@ -2451,7 +2451,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
// Need to do these checks inside the synchronized
if (paused || !canDispatch() && redistributor == null) {
- return;
+ return false;
}
if (messageReferences.size() == 0) {
@@ -2571,7 +2571,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
}
}
- checkDepage();
+ return true;
}
protected void removeMessageReference(ConsumerHolder<? extends Consumer> holder, MessageReference ref) {
@@ -3423,13 +3423,19 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
// We will be using the deliverRunner instance as the guard object to avoid multiple threads executing
// an asynchronous delivery
enterCritical(CRITICAL_DELIVER);
+ boolean needCheckDepage = false;
try {
synchronized (QueueImpl.this.deliverRunner) {
- deliver();
+ needCheckDepage = deliver();
}
} finally {
leaveCritical(CRITICAL_DELIVER);
}
+
+ if (needCheckDepage) {
+ checkDepage();
+ }
+
} catch (Exception e) {
ActiveMQServerLogger.LOGGER.errorDelivering(e);
} finally {