Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-2244 checkDepage method placed outside CRITICAL_DELIVER avoid critical analyzer timeout

This commit is contained in:
yb 2019-01-31 17:20:06 +08:00 committed by Clebert Suconic
parent 4f758f3dd9
commit 8799615a13
1 changed files with 12 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
View File

@ -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 {