When consumer not finishes execution within SimpleMessageListenerContainer shutdownTimeout after AnnotationConfigApplicationContext.close() invocation, my Spring application hangs.
Listener:
public class LongRunningMessageListener implements MessageListener {
@Override
public void onMessage(Message message) {
System.out.println("Got " + new String(message.getBody()));
try {
Thread.sleep(5000L);
} catch (InterruptedException e) {
System.out.println("Interrupted");
Thread.currentThread().interrupt();
}
System.out.println("Finished execution");
}
}
Configuration:
@Configuration
public class Config {
@Bean
public MessageListenerContainer messageListenerContainer() {
SimpleMessageListenerContainer messageListenerContainer = new SimpleMessageListenerContainer(connectionFactory());
messageListenerContainer.setQueueNames("myqueue");
messageListenerContainer.setMessageListener(new LongRunningMessageListener());
messageListenerContainer.setShutdownTimeout(1000);
return messageListenerContainer;
}
@Bean
public ConnectionFactory connectionFactory() {
return new CachingConnectionFactory("localhost");
}
}
Main:
public static void main(String[] args) throws InterruptedException {
AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext(Config.class);
Thread.sleep(1000L);
applicationContext.close();
}
Output:
мая 08, 2015 3:43:21 PM org.springframework.context.annotation.AnnotationConfigApplicationContext prepareRefresh
INFO: Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@1a86f2f1: startup date [Fri May 08 15:43:21 MSK 2015]; root of context hierarchy
мая 08, 2015 3:43:22 PM org.springframework.context.support.DefaultLifecycleProcessor start
INFO: Starting beans in phase 2147483647
мая 08, 2015 3:43:22 PM org.springframework.amqp.rabbit.connection.CachingConnectionFactory createBareConnection
INFO: Created new connection: SimpleConnection@4025f2f5 [delegate=amqp://guest@127.0.0.1:5672/]
Got foo1
мая 08, 2015 3:43:23 PM org.springframework.context.annotation.AnnotationConfigApplicationContext doClose
INFO: Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@1a86f2f1: startup date [Fri May 08 15:43:21 MSK 2015]; root of context hierarchy
мая 08, 2015 3:43:23 PM org.springframework.context.support.DefaultLifecycleProcessor stop
INFO: Stopping beans in phase 2147483647
мая 08, 2015 3:43:23 PM org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer doShutdown
INFO: Waiting for workers to finish.
мая 08, 2015 3:43:24 PM org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer doShutdown
INFO: Workers not finished. Forcing connections to close.
Finished execution
мая 08, 2015 3:43:27 PM org.springframework.amqp.rabbit.connection.CachingConnectionFactory createBareConnection
INFO: Created new connection: SimpleConnection@260da64e [delegate=amqp://guest@127.0.0.1:5672/]
мая 08, 2015 3:43:27 PM org.springframework.amqp.rabbit.connection.CachingConnectionFactory shutdownCompleted
SEVERE: Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)
After that application is working infinitely.
Working threads:
Live threads: 14
Daemon threads: 10
Non daemon threads:
"pool-1-thread-10" #22 prio=5 os_prio=31 tid=0x00007fd48c93c800 nid=0x5a07 waiting on condition [0x0000000129eb3000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x0000000796333168> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Locked ownable synchronizers:
- None
"pool-1-thread-9" #21 prio=5 os_prio=31 tid=0x00007fd48b9d8000 nid=0x540b waiting on condition [0x0000000129db0000]
java.lang.Thread.State: TIMED_WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x000000079633dfb0> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Locked ownable synchronizers:
- None
"AMQP Connection 127.0.0.1:5672" #20 prio=5 os_prio=31 tid=0x00007fd48d186000 nid=0x530b runnable [0x0000000129cad000]
java.lang.Thread.State: RUNNABLE
at java.net.SocketInputStream.socketRead0(Native Method)
at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
at java.net.SocketInputStream.read(SocketInputStream.java:170)
at java.net.SocketInputStream.read(SocketInputStream.java:141)
at java.io.BufferedInputStream.fill(BufferedInputStream.java:246)
at java.io.BufferedInputStream.read(BufferedInputStream.java:265)
- locked <0x000000079632e8f0> (a java.io.BufferedInputStream)
at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:288)
at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:95)
at com.rabbitmq.client.impl.SocketFrameHandler.readFrame(SocketFrameHandler.java:139)
- locked <0x000000079632e8d0> (a java.io.DataInputStream)
at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:534)
at java.lang.Thread.run(Thread.java:745)
Locked ownable synchronizers:
- None
"DestroyJavaVM" #19 prio=5 os_prio=31 tid=0x00007fd48d002000 nid=0x1303 waiting on condition [0x0000000000000000]
java.lang.Thread.State: RUNNABLE
Locked ownable synchronizers:
- None
Everything is fine if I increase shutdownTimeout, but timeout behavior looks like a bug.
Spring versions:
<dependencies>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-amqp</artifactId>
<version>1.4.5.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.4.5.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>4.1.5.RELEASE</version>
</dependency>
</dependencies>
Aucun commentaire:
Enregistrer un commentaire