一.DefaultMessageListenerContainer概述
DefaultMessageListenerContainer是一个用于异步消息监听的管理类。
DefaultMessageListenerContainer最简单的实现逻辑,一个任务执行器,执行任务(即消息监听)。
DefaultMessageListenerContainer实现的主要原理是,通过内部初始化建立的一个taskExecutor(默认是SimpleAsyncTaskExecutor)用于执行消息监听的任务(AsyncMessageListenerInvoker)。
这里默认的任务执行器是SimpleAsyncTaskExecutor,这个执行器的缺点是不会重用连接,也就是对于每个任务都需要新开启一个线程,执行完任务后会关闭它。如果要优化的话可以考虑线程池。
消息监听的任务被抽象成AsyncMessageListenerInvoker类,这个类实现了Runnable接口,内部run方法其实是通过不断循环consumer.recieve()方法来实现监听。
事实上一个消费者对应了一个AsyncMessageListenerInvoker任务,每个任务需要一个单独的线程去执行它。这个AsyncMessageListenerInvoker实例被放在了一个名为scheduledInvokers的set里面。
二.DefaultMessageListenerContainer中的connection,session,consumer
其实我们还有一个比较关心的地方是这个DefaultMessageListenerContainer缓不缓存connection、session、consumer。它是根据catchLevel属性来决定是否缓存connection、session、consumer。默认的catchLevel对应常量CATCH_AUTO,即由配置的外部事务管理器决定。catchLevel级别分别是CATCH_NONE,CATCH_CONNECTION,CATCH_SESSION,CATCH_CONSUMER,分别对应0,1,2,3。我试了下默认的CATCH_AUTO在没有定义事务管理时值为 CATCH_CONSUMER,即3。具体查看类中的方法:
/** * Constant that indicates to cache no JMS resources at all. * @see #setCacheLevel */ public static final int CACHE_NONE = 0; /** * Constant that indicates to cache a shared JMS {@code Connection} for each * listener thread. * @see #setCacheLevel */ public static final int CACHE_CONNECTION = 1; /** * Constant that indicates to cache a shared JMS {@code Connection} and a JMS * {@code Session} for each listener thread. * @see #setCacheLevel */ public static final int CACHE_SESSION = 2; /** * Constant that indicates to cache a shared JMS {@code Connection}, a JMS * {@code Session}, and a JMS MessageConsumer for each listener thread. * @see #setCacheLevel */ public static final int CACHE_CONSUMER = 3; /** * Constant that indicates automatic choice of an appropriate caching level * (depending on the transaction management strategy). * @see #setCacheLevel */ public static final int CACHE_AUTO = 4;
@Override public void initialize() { // Adapt default cache level. if (this.cacheLevel == CACHE_AUTO) { this.cacheLevel = (getTransactionManager() != null ? CACHE_NONE : CACHE_CONSUMER); } // Prepare taskExecutor and maxMessagesPerTask. synchronized (this.lifecycleMonitor) { if (this.taskExecutor == null) { this.taskExecutor = createDefaultTaskExecutor(); } else if (this.taskExecutor instanceof SchedulingTaskExecutor && ((SchedulingTaskExecutor) this.taskExecutor).prefersShortLivedTasks() && this.maxMessagesPerTask == Integer.MIN_VALUE) { // TaskExecutor indicated a preference for short-lived tasks. According to // setMaxMessagesPerTask javadoc, we'll use 10 message per task in this case // unless the user specified a custom value. this.maxMessagesPerTask = 10; } } // Proceed with actual listener initialization. super.initialize(); }
DefaultMessageListenerContainer会根据catchLevel来缓存共享connection、session及consumer,值为3的话就会缓存connection、session及consumer,在初始化的时候就会调用父类AbstractJmsListeningContainer的doStart()方法,判断cacheLevel是否大于等于1,如果大于就创建一个connection将放入成员变量sharedConnection中。
每个任务被执行的时候(即责任是监听消息),会先去获取connection、session及consumer(通过调用initResourcesIfNecessary方法)就像我们自己最初实现一个简单的客户端消费者一样。只不过这里会根据catchLevel来决定是否缓存session及consumer,被缓存了的session及consumer放在对应的成员变量里面。
接着任务会想要执行consumer.recieve方法,这之前肯定要获取onnection、session及consumer,如果已有connection、session及consumer则获取过来,如果没有则通过配置的信息新建。
执行完consumer.recieve后,会判断consumer.recieve返回的消息是否为空,不为空则调用message对应的messageListner(之前我们在DefaultMessageListenerContainer中通过方法setMessageListner设置的)的onMessage执行相应的逻辑,并设置这个任务的Idle为false,表明这个任务不是空闲的,然后会调用方法判断是否应该新建任务实例,这个受限于MaxConcurrentConsumers及IdleTaskExecutionLimit。为空则不需要特别处理,只需调用noMessageReceived方法将idle标记设为true。
任务执行完后,会在finally处释放connection,session及consumer。这个是根据上述讲的catchLevel来设置的。
继承体系如下:
AbstractJmsListeningContainer提供了一个最上层最基础的jms消息监听管理类所应该有的方法。提供了start(启动这个管理类)、stop、initialize(初始化这个管理类)、establishSharedConnection等。
三.DefaultMessageListenerContainer类的主要属性
DefaultMessageListenerContainer继承自AbstractPollingMessageListenerContainer,主要使用同步的方式接收消息(也就是通过循环调用MessageConsumer.receive的方式接收消息)。
DefaultMessageListenerContainer类主要的属性如下:
private Executor taskExecutor; private long recoveryInterval = DEFAULT_RECOVERY_INTERVAL; private int cacheLevel = CACHE_AUTO; private int concurrentConsumers = 1; private int maxConcurrentConsumers = 1; private int maxMessagesPerTask = Integer.MIN_VALUE; private int idleConsumerLimit = 1; private int idleTaskExecutionLimit = 1; private final Set<AsyncMessageListenerInvoker> scheduledInvokers = new HashSet<AsyncMessageListenerInvoker>();
跟SimpleMessageListenerContainer一样,DefaultMessageListenerContainer也支持创建多个Session和MessageConsumer来接收消息。跟SimpleMessageListenerContainer不同的是,DefaultMessageListenerContainer创建了concurrentConsumers所指定个数的AsyncMessageListenerInvoker(实现了SchedulingAwareRunnable接口),并交给taskExecutor运行。
maxMessagesPerTask属性的默认值是Integer.MIN_VALUE,但是如果设置的taskExecutor(默认值是SimpleAsyncTaskExecutor)实现了SchedulingTaskExecutor接口并且其prefersShortLivedTasks方法返回true(也就是说该TaskExecutor倾向于短期任务),那么maxMessagesPerTask属性会自动被设置为10。
如果maxMessagesPerTask属性的值小于0,那么AsyncMessageListenerInvoker.run方法会在循环中反复尝试接收消息,并在接收到消息后调用MessageListener(或者SessionAwareMessageListener);如果maxMessagesPerTask属性的值不小于0,那么AsyncMessageListenerInvoker.run方法里最多会尝试接收消息maxMessagesPerTask次,每次接收消息的超时时间由其父类AbstractPollingMessageListenerContainer的receiveTimeout属性指定。如果在这些尝试中都没有接收到消息,那么AsyncMessageListenerInvoker的idleTaskExecutionCount属性会被累加。在run方法执行完毕前会对idleTaskExecutionCount进行检查,如果该值超过了DefaultMessageListenerContainer.idleTaskExecutionLimit(默认值1),那么这个AsyncMessageListenerInvoker可能会被销毁。
所有AsyncMessageListenerInvoker实例都保存在scheduledInvokers中,实例的个数可以在concurrentConsumers和maxConcurrentConsumers之间浮动。跟SimpleMessageListenerContainer一样,应该只是在Destination为Queue的时候才使用多个AsyncMessageListenerInvoker实例。
cacheLevel属性用于指定是否对JMS资源进行缓存,可选的值是CACHE_NONE = 0、CACHE_CONNECTION = 1、CACHE_SESSION = 2、CACHE_CONSUMER = 3和CACHE_AUTO = 4。默认情况下,如果transactionManager属性不为null,那么cacheLevel被自动设置为CACHE_NONE(不进行缓存),否则cacheLevel被自动设置为CACHE_CONSUMER。
如果cacheLevel属性值大于等于CACHE_CONNECTION,那么sharedConnectionEnabled方法(在AbstractJmsListeningContainer中定义)返回true,也就是说使用共享的JMS连接。
四.设置messageSelector
DefaultMessageListenerContainer是支持动态改变messageSelector的,可以动态设置messageSelector,Container就能用上最新的selector了。
Spring配置如下:
<bean id="messageListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="jmsConnectionFactory" /> <property name="destination" ref="receiverQueue" /> <property name="messageListener" ref="jmsReceiver" /> <property name="concurrentConsumers" value="10" /> <property name="messageSelector" value="CLIENT='DEMO'" /> <property name="cacheLevel" value="2"/> </bean>
修改messageSelector代码如下:
DefaultMessageListenerContainer messageListenerContainer = (DefaultMessageListenerContainer) ac.getBean("messageListenerContainer"); messageListenerContainer.setMessageSelector("CLIENT='DEMO2'");
五.源码分析
1.DefaultMessageListenerContainer类中的invokeListener方法,每次收消息都会调用。
private boolean invokeListener() throws JMSException { initResourcesIfNecessary(); boolean messageReceived = receiveAndExecute(this, this.session, this.consumer); this.lastMessageSucceeded = true; return messageReceived; }
/** * Runnable that performs looped {@code MessageConsumer.receive()} calls. */ private class AsyncMessageListenerInvoker implements SchedulingAwareRunnable { private Session session; private MessageConsumer consumer; private Object lastRecoveryMarker; private boolean lastMessageSucceeded; private int idleTaskExecutionCount = 0; private volatile boolean idle = true; @Override public void run() { synchronized (lifecycleMonitor) { activeInvokerCount++; lifecycleMonitor.notifyAll(); } boolean messageReceived = false; try { if (maxMessagesPerTask < 0) { messageReceived = executeOngoingLoop(); } else { int messageCount = 0; while (isRunning() && messageCount < maxMessagesPerTask) { messageReceived = (invokeListener() || messageReceived); messageCount++; } } } catch (Throwable ex) { clearResources(); if (!this.lastMessageSucceeded) { // We failed more than once in a row - sleep for recovery interval // even before first recovery attempt. sleepInbetweenRecoveryAttempts(); }
2.如下initResourcesIfNecessary方法中使用cacheLevel的地方,由于需要动态selector,所以需要每次重新生成consumer,当cacheLevel<3的时候,this.consumer会为null。
private void initResourcesIfNecessary() throws JMSException { if (getCacheLevel() <= CACHE_CONNECTION) { updateRecoveryMarker(); } else { if (this.session == null && getCacheLevel() >= CACHE_SESSION) { updateRecoveryMarker(); this.session = createSession(getSharedConnection()); } if (this.consumer == null && getCacheLevel() >= CACHE_CONSUMER) { this.consumer = createListenerConsumer(this.session); synchronized (lifecycleMonitor) { registeredWithDestination++; } } } }
3.在AbstractPollingMessageListenerContainer的doReceiveAndExecute方法中可以发现当传入consumer为null时,会生成一个新的consumer。
/** * Actually execute the listener for a message received from the given consumer, * fetching all requires resources and invoking the listener. * @param session the JMS Session to work on * @param consumer the MessageConsumer to work on * @param status the TransactionStatus (may be {@code null}) * @return whether a message has been received * @throws JMSException if thrown by JMS methods * @see #doExecuteListener(javax.jms.Session, javax.jms.Message) */ protected boolean doReceiveAndExecute( Object invoker, Session session, MessageConsumer consumer, TransactionStatus status) throws JMSException { Connection conToClose = null; Session sessionToClose = null; MessageConsumer consumerToClose = null; try { Session sessionToUse = session; boolean transactional = false; if (sessionToUse == null) { sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession( getConnectionFactory(), this.transactionalResourceFactory, true); transactional = (sessionToUse != null); } if (sessionToUse == null) { Connection conToUse; if (sharedConnectionEnabled()) { conToUse = getSharedConnection(); } else { conToUse = createConnection(); conToClose = conToUse; conToUse.start(); } sessionToUse = createSession(conToUse); sessionToClose = sessionToUse; } MessageConsumer consumerToUse = consumer; if (consumerToUse == null) { consumerToUse = createListenerConsumer(sessionToUse); consumerToClose = consumerToUse; } Message message = receiveMessage(consumerToUse); if (message != null) { if (logger.isDebugEnabled()) { logger.debug("Received message of type [" + message.getClass() + "] from consumer [" + consumerToUse + "] of " + (transactional ? "transactional " : "") + "session [" + sessionToUse + "]"); } messageReceived(invoker, sessionToUse); boolean exposeResource = (!transactional && isExposeListenerSession() && !TransactionSynchronizationManager.hasResource(getConnectionFactory())); if (exposeResource) { TransactionSynchronizationManager.bindResource( getConnectionFactory(), new LocallyExposedJmsResourceHolder(sessionToUse)); } try { doExecuteListener(sessionToUse, message); } catch (Throwable ex) { if (status != null) { if (logger.isDebugEnabled()) { logger.debug("Rolling back transaction because of listener exception thrown: " + ex); } status.setRollbackOnly(); } handleListenerException(ex); // Rethrow JMSException to indicate an infrastructure problem // that may have to trigger recovery... if (ex instanceof JMSException) { throw (JMSException) ex; } } finally { if (exposeResource) { TransactionSynchronizationManager.unbindResource(getConnectionFactory()); } } // Indicate that a message has been received. return true; } else { if (logger.isTraceEnabled()) { logger.trace("Consumer [" + consumerToUse + "] of " + (transactional ? "transactional " : "") + "session [" + sessionToUse + "] did not receive a message"); } noMessageReceived(invoker, sessionToUse); // Nevertheless call commit, in order to reset the transaction timeout (if any). // However, don't do this on Tibco since this may lead to a deadlock there. if (shouldCommitAfterNoMessageReceived(sessionToUse)) { commitIfNecessary(sessionToUse, message); } // Indicate that no message has been received. return false; } } finally { JmsUtils.closeMessageConsumer(consumerToClose); JmsUtils.closeSession(sessionToClose); ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), true); } }
文章来源:http://blog.csdn.net/caolaosanahnu/article/details/12096577