主要的几个类说明:
1 JmsTemplate 用于发送和接受消息。需要消息工厂参数。
基于监听：
2 DefaultMessageListenerContainer，SimpleMessageListenerContainer这两个容器可以创建多个session和消费者来对每个队列进行消息处理并条用消息监听类的方法进行处理。并通过多线程进行处理。每个线程通过轮训的方式(while(true))去进行消息接收处理。
3 MessageListenerAdapter 它相当于一个特殊的自定义监听器，里面可以调用一些消息格式的装换工具。例如json，xml，string到bean的转换等等。
其中DefaultMessageListenerContainer接受动态添加session。而后者不支持。
----
具体消息监听处理源码过程如下：以DefaultMessageListenerContainer为例：
----
1 初始化建立消费者线程：
/**
* Creates the specified number of concurrent consumers,
* in the form of a JMS Session plus associated MessageConsumer.
* @see #createListenerConsumer
*/
protected void doInitialize() throws JMSException {
establishSharedConnection();
initializeConsumers();
}
protected void initializeConsumers() throws JMSException {
// Register Sessions and MessageConsumers.
synchronized (this.consumersMonitor) {
if (this.consumers == null) {
this.sessions = new HashSet<Session>(this.concurrentConsumers);
this.consumers = new HashSet<MessageConsumer>(this.concurrentConsumers);
Connection con = getSharedConnection();
for (int i = 0; i < this.concurrentConsumers; i++) {
Session session = createSession(con);
MessageConsumer consumer = createListenerConsumer(session);
this.sessions.add(session);
this.consumers.add(consumer);
}
}
}
}
2 启动消费者线程
/**
* Start the shared Connection, if any, and notify all invoker tasks.
* @throws JMSException if thrown by JMS API methods
* @see #startSharedConnection
*/
protected void doStart() throws JMSException {
// Lazily establish a shared Connection, if necessary.
if (sharedConnectionEnabled()) {
establishSharedConnection();
}
// Reschedule paused tasks, if any.
synchronized (this.lifecycleMonitor) {
this.running = true;
this.lifecycleMonitor.notifyAll();
resumePausedTasks();
}
// Start the shared Connection, if any.
if (sharedConnectionEnabled()) {
startSharedConnection();
}
}
3消息消费
其中的resumePausedTasks方法会进行轮训处理：
protected void resumePausedTasks() {
synchronized (this.lifecycleMonitor) {
if (!this.pausedTasks.isEmpty()) {//所有的线程一开始会被置为暂停
for (Iterator<?> it = this.pausedTasks.iterator(); it.hasNext();) {
Object task = it.next();
try {
doRescheduleTask(task);
it.remove();
if (logger.isDebugEnabled()) {
logger.debug("Resumed paused task: " + task);
}
}
catch (RuntimeException ex) {
logRejectedTask(task, ex);
// Keep the task in paused mode...
}
}
}
}
每个线程是一个 AsyncMessageListenerInvoker，通过它的run方法来进行消息处理：
先执行：
private boolean invokeListener() throws JMSException {
initResourcesIfNecessary();
boolean messageReceived = receiveAndExecute(this, this.session, this.consumer);
this.lastMessageSucceeded = true;
return messageReceived;
}
接着：
Message message = receiveMessage(consumerToUse);
这就是最终的阻塞方法去获取消息：
/**
* Receive a message from the given consumer.
* @param consumer the MessageConsumer to use
* @return the Message, or <code>null</code> if none
* @throws JMSException if thrown by JMS methods
*/
protected Message receiveMessage(MessageConsumer consumer) throws JMSException {
return (this.receiveTimeout < 0 ? consumer.receive() : consumer.receive(this.receiveTimeout));
}
4调用我们自定义的监听器：
其中监听器可以实现SessionAwareMessageListener和MessageListener接口，不一样的地方在于前者会把session，message参数给你，你可以进行一个回调操作。后者只有消息内容message参数。
protected void invokeListener(Session session, Message message) throws JMSException {
Object listener = getMessageListener();
if (listener instanceof SessionAwareMessageListener) {
doInvokeListener((SessionAwareMessageListener) listener, session, message);
}
else if (listener instanceof MessageListener) {
doInvokeListener((MessageListener) listener, message);
}
else if (listener != null) {
throw new IllegalArgumentException(
"Only MessageListener and SessionAwareMessageListener supported: " + listener);
}
else {
throw new IllegalStateException("No message listener specified - see property 'messageListener'");
}
}