【spring-jms】DefaultMessageListenerContainer

2019-11-11 02:51:00

字体：大中小

来源：转载

供稿：网友

DefaultMessageListenerContainer 分析

org.sPRingframework.jms.listener.DefaultMessageListenerContainer

通过分析该源码，了解下 Spring 下消费者如何进行管理。。。

maven

<dependency> <groupId>org.springframework</groupId> <artifactId>spring-jms</artifactId> <version>4.1.4.RELEASE</version></dependency>

重要步骤

初始化 initialize()默认的线程执行器 SimpleAsyncTaskExecutor, createDefaultTaskExecutor()恢复策略 recoverAfterListenerSetupFailure()

调用栈

先看 Refer 1 中的一个配置

<bean id="queueDestination2" class="org.apache.activemq.command.ActiveMQQueue">  <constructor-arg> <value>queue2</value> </constructor-arg></bean><bean id="queueMessageListener" class="guo.examples.mq02.queue.QueueMessageListener" /><bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="connectionFactory" /> <property name="destination" ref="queueDestination2" /> <property name="messageListener" ref="queueMessageListener" /></bean>

下面是 messageListener 的调用栈

AsyncMessageListenerInvoker.run()↓AsyncMessageListenerInvoker.invokeListener()↓AbstractPollingMessageListenerContainer.receiveAndExecute(Object invoker, session session, MessageConsumer consumer)↓AbstractPollingMessageListenerContainer.doReceiveAndExecute(Object invoker, Session session, MessageConsumer consumer, TransactionStatus status)↓AbstractMessageListenerContainer.doExecuteListener(Session session, Message message)# 判断 isAcceptMessagesWhileStopping & isRunning↓AbstractMessageListenerContainer.invokeListener(Session session, Message message)↓# @param listener 即上面用户定义的 MessageListener# 由于 SimpleAsyncTaskExecutor 是内部类，所以可以访问 DefaultMessageListenerContainer.messageListenerAbstractMessageListenerContainer.doInvokeListener(MessageListener listener, Message message)↓MessageListener.onMessage(Message message)

SimpleAsyncTaskExecutor 线程执行过程

# 创建新线程DefaultMessageListenerContainer.scheduleNewInvoker()↓# 判断 DefaultMessageListenerContainer 处于 active, running 哪一状态，进行后面的操作AbstractJmsListeningContainer.rescheduleTaskIfNecessary(Object task)# 如果 running↓DefaultMessageListenerContainer.doRescheduleTask(Object task)# 注意，不是 AbstractJmsListeningContainer.doRescheduleTask(Object task)↓# 然后执行上面的 `调用栈` 过程