深入理解DefaultMessageListenerContainer -

bijian1013

浏览: 3186331 次
性别:
来自: 深圳

最近访客更多访客>>

dy.f

godandghost

lovefiger

marsgrayson

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

深入理解DefaultMessageListenerContainer

博客分类：

WebSphere MQ

DefaultMessageListenerContainer JMS

一.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

http://kooii.iteye.com/blog/1913272