【spring-jms】DefaultMessageListenerContainer

2019-11-11 02:51:00

字體：大中小

供稿：網(wǎng)友

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()默認的線程執(zhí)行器 SimpleAsyncTaskExecutor, createDefaultTaskExecutor()恢復(fù)策略 recoverAfterListenerSetupFailure()

調(diào)用棧

先看 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 的調(diào)用棧

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 是內(nèi)部類，所以可以訪問 DefaultMessageListenerContainer.messageListenerAbstractMessageListenerContainer.doInvokeListener(MessageListener listener, Message message)↓MessageListener.onMessage(Message message)

SimpleAsyncTaskExecutor 線程執(zhí)行過程

# 創(chuàng)建新線程DefaultMessageListenerContainer.scheduleNewInvoker()↓# 判斷 DefaultMessageListenerContainer 處于 active, running 哪一狀態(tài)，進行后面的操作AbstractJmsListeningContainer.rescheduleTaskIfNecessary(Object task)# 如果 running↓DefaultMessageListenerContainer.doRescheduleTask(Object task)# 注意，不是 AbstractJmsListeningContainer.doRescheduleTask(Object task)↓# 然后執(zhí)行上面的 `調(diào)用棧` 過程