Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Interrupt listener invoker threads on shutdown (after initial wait step)

Browse Source 
Issue: SPR-16536
This commit is contained in:
Juergen Hoeller
2018-02-28 00:06:46 +01:00
parent d1ccecd021
commit 95aad9cdc2
1 changed files with 38 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
spring-jms/src/main/java/org/springframework/jms/listener/DefaultMessageListenerContainer.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2016 the original author or authors. * Copyright 2002-2018 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -563,21 +563,32 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
logger.debug("Waiting for shutdown of message listener invokers"); logger.debug("Waiting for shutdown of message listener invokers");
try { try {
synchronized (this.lifecycleMonitor) { synchronized (this.lifecycleMonitor) {
// Waiting for AsyncMessageListenerInvokers to deactivate themselves... long receiveTimeout = getReceiveTimeout();
long waitStartTime = System.currentTimeMillis();
int waitCount = 0;
while (this.activeInvokerCount > 0) { while (this.activeInvokerCount > 0) {
if (waitCount > 0 && !isAcceptMessagesWhileStopping() &&
System.currentTimeMillis() - waitStartTime >= receiveTimeout) {
// Unexpectedly some invokers are still active after the receive timeout period
// -> interrupt remaining receive attempts since we'd reject the messages anyway
for (AsyncMessageListenerInvoker scheduledInvoker : this.scheduledInvokers) {
scheduledInvoker.interruptIfNecessary();
}
}
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("Still waiting for shutdown of " + this.activeInvokerCount + logger.debug("Still waiting for shutdown of " + this.activeInvokerCount +
" message listener invokers"); " message listener invokers (iteration " + waitCount + ")");
} }
long timeout = getReceiveTimeout(); // Wait for AsyncMessageListenerInvokers to deactivate themselves...
if (timeout > 0) { if (receiveTimeout > 0) {
this.lifecycleMonitor.wait(timeout); this.lifecycleMonitor.wait(receiveTimeout);
} }
else { else {
this.lifecycleMonitor.wait(); this.lifecycleMonitor.wait();
} }
waitCount++;
} }
// Clear remaining scheduled invokers, possibly left over as paused tasks... // Clear remaining scheduled invokers, possibly left over as paused tasks
for (AsyncMessageListenerInvoker scheduledInvoker : this.scheduledInvokers) { for (AsyncMessageListenerInvoker scheduledInvoker : this.scheduledInvokers) {
scheduledInvoker.clearResources(); scheduledInvoker.clearResources();
} }
@@ -1050,6 +1061,9 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
private volatile boolean idle = true; private volatile boolean idle = true;
@Nullable
private volatile Thread currentReceiveThread;
@Override @Override
public void run() { public void run() {
synchronized (lifecycleMonitor) { synchronized (lifecycleMonitor) {
@@ -1169,10 +1183,16 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
} }
private boolean invokeListener() throws JMSException { private boolean invokeListener() throws JMSException {
initResourcesIfNecessary(); this.currentReceiveThread = Thread.currentThread();
boolean messageReceived = receiveAndExecute(this, this.session, this.consumer); try {
this.lastMessageSucceeded = true; initResourcesIfNecessary();
return messageReceived; boolean messageReceived = receiveAndExecute(this, this.session, this.consumer);
this.lastMessageSucceeded = true;
return messageReceived;
}
finally {
this.currentReceiveThread = null;
}
} }
private void decreaseActiveInvokerCount() { private void decreaseActiveInvokerCount() {
@@ -1207,6 +1227,13 @@ public class DefaultMessageListenerContainer extends AbstractPollingMessageListe
} }
} }
private void interruptIfNecessary() {
Thread currentReceiveThread = this.currentReceiveThread;
if (currentReceiveThread != null && !currentReceiveThread.isInterrupted()) {
currentReceiveThread.interrupt();
}
}
private void clearResources() { private void clearResources() {
if (sharedConnectionEnabled()) { if (sharedConnectionEnabled()) {
synchronized (sharedConnectionMonitor) { synchronized (sharedConnectionMonitor) {