1 public class Sender{
 2     public static void main(String[] args) throws Exception{
 3         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
 4         Connection connection = connectionFactory.createConnection();
 5         connection.start()
 6 
 7         Session session = connection.creatSession(Boolean.True, Session.AUTO_ACKNOWLEDGE);
 8         Destination destination = session.createQueue("my-queue");
 9 
10         MessageProducer producer = session.createProducer(destionation);
11         for (int i=0; i<3; i++){
12             TextMessage message = session.createTextMessage("大家好，這是個測試")；
13             Tread.sleep(1000);
14             //通過消息生產者發出消息
15             producer.send(message);
16         }
17         session.commit();
18         session.close();
19         connection.close();
20     }
21 }

 1 public class Receiver{
 2     public static void main(String[] args) throws Exception {
 3         ConnectionFactory connectionFactory = new ActiveMQConnectionFactroy();
 4         Connection connection = connectionFactory.createConnection();
 5         connection.start();
 6 
 7         final Session session = connection.createSession(Boolean.TRUE, Session.AUTOACKNOWLEDGE);
 8         Destination destination = session.createQueue("my-queue");
 9         MessageConsumer consumer = session.createConsumer(destination);
10 
11         int i = 0;
12         while(i<3){
13             i++;
14             TextMessage message = (TextMessage) consumer.receive();
15             session.commit();
16             //TODO something...
17             System.out.println("收到消息：" + message.getText());
18         }
19 
20         session.close();
21         connection.close();
22     }
23 }

 1 <beans>
 2     <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
 3         <Property name="brokerURL">
 4             <value>tcp://localhost:61616</value>
 5         </Property>
 6     </bean>
 7 
 8     <bean id="jmsTemplate" class="org.Springframework.jms.core.JmsTemplate">
 9         <Property name="connectionFactory">
10             <ref bean="connectionFactory" />
11         </Property>
12     </bean>
13 
14     <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
15         <constructor-arg index="0">
16             <vaule>HelloWordQueue</vaule>
17         </constructor-arg>
18     </bean>
19 
20 </beans>

 1 public class HelloWorldSender{
 2     public static void main(String[] args) throws Exception{
 3         ApplicationContext context = new ClassPathXmlApplicationContext(new string[] {"test/activeMQ/Spring/applicationContext.xml"});
 4 
 5         JmsTemplate jmsTemplate = (JmsTemplate) context.getBean("jmsTemplate");
 6         Destination destination = (Destination) context.getBean("destination");
 7 
 8         jmsTemplate.send(destination, new MessageCreator(){
 9             public Message createMessage(Session session) throws JMSException{
10                 return session.createTextMessage("大家好，這是個測試");
11             }
12         });
13     }
14 }

 1 public class HelloWorldReceiver{
 2     public static void main(string[] args) throws Exception{
 3         ApplicationContext context = new ClassPathXMLApplicationContext(new String[] {"test/activeMQ/Spring/applicationContext.xml"});
 4         JmsTemplate jmsTemplate = (JmsTemplate) context.getBean("jmsTemplate");
 5         Destination destination = (Destination) context.getBean("destination");
 6 
 7         TextMessage msg = (TextMessage) jmsTemplate.receive(destination);
 8         System.out.println("received msg is:" + msg.getText());
 9     }
10 }

 1 public class MyMessageListener implements MessageListener{
 2 
 3     @Override
 4     public void onMessage(Message arg0){
 5         TextMessage msg = (TextMessage) arg0;
 6         try{
 7             System.out.println(msg.getText());
 8         } catch (JMSException e){
 9             e.printStackTrace();
10         }
11     }
12 }

 1 <beans>
 2     <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
 3         <Property name="brokerURL">
 4             <value>tcp://localhost:61616</value>
 5         </Property>
 6     </bean>
 7 
 8     <bean id="jmsTemplate" class="org.Springframework.jms.core.JmsTemplate">
 9         <Property name="connectionFactory">
10             <ref bean="connectionFactory" />
11         </Property>
12     </bean>
13 
14     <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
15         <constructor-arg index="0">
16             <vaule>HelloWordQueue</vaule>
17         </constructor-arg>
18     </bean>
19 
20     <bean id="myTextListener" class="test.activeMQ.Spring.MyMessageListener" />
21 
22     <bean id="javaConsumer" class="org.Springframework.jms.listener.DefaultMessageListenerContainer">
23         <property name="ConnectionFactory" ref="connectionFactory" />
24         <Property name="destination" ref="destination" />
25         <Property name="messageListener" ref="myTextListener" />
26     </bean>
27 
28 </beans>

1 jmsTemplate.send(destination, new MessageCreator(){
2     public Message createMessage(Session session) throws JMSException{
3         return session.createTextMessage("大家好，這是個測試");
4     }
5 });

1     public void send(final Destination destination, final MessageCreator messageCreator) throws JmsException {
2         execute(session -> {
3             doSend(session, destination, messageCreator);
4             return null;
5         }, false);
6     }

 1     public <T> T execute(SessionCallback<T> action, boolean startConnection) throws JmsException {
 2         Assert.notNull(action, "Callback object must not be null");
 3         Connection conToClose = null;
 4         Session sessionToClose = null;
 5         try {
 6             Session sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession(
 7                     obtainConnectionFactory(), this.transactionalResourceFactory, startConnection);
 8             if (sessionToUse == null) {
 9                 //創建connection
10                 conToClose = createConnection();
11                 //根據connection創建session
12                 sessionToClose = createSession(conToClose);
13                 //是否開啟向服務推送連接信息，只有接收信息時需要，發送時不需要
14                 if (startConnection) {
15                     conToClose.start();
16                 }
17                 sessionToUse = sessionToClose;
18             }
19             if (logger.isDebugEnabled()) {
20                 logger.debug("Executing callback on JMS Session: " + sessionToUse);
21             }
22             //調用回調函數
23             return action.doInJms(sessionToUse);
24         }
25         catch (JMSException ex) {
26             throw convertJmsAccessException(ex);
27         }
28         finally {
29             //關閉session
30             JmsUtils.closeSession(sessionToClose);
31             //釋放連接
32             ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), startConnection);
33         }
34     }

1 new SessionCallback<Object>(){
2    public Object doInJms(Session session) throws JMSException {
3       doSend(session, destination, messageCreator);
4       return null;
5    }
6 }

 1     protected void doSend(Session session, Destination destination, MessageCreator messageCreator)
 2             throws JMSException {
 3 
 4         Assert.notNull(messageCreator, "MessageCreator must not be null");
 5         MessageProducer producer = createProducer(session, destination);
 6         try {
 7             Message message = messageCreator.createMessage(session);
 8             if (logger.isDebugEnabled()) {
 9                 logger.debug("Sending created message: " + message);
10             }
11             doSend(producer, message);
12             // Check commit - avoid commit call within a JTA transaction.
13             if (session.getTransacted() && isSessionLocallyTransacted(session)) {
14                 // Transacted session created by this template -> commit.
15                 JmsUtils.commitIfNecessary(session);
16             }
17         }
18         finally {
19             JmsUtils.closeMessageProducer(producer);
20         }
21     }

 1     protected void doSend(MessageProducer producer, Message message) throws JMSException {
 2         if (this.deliveryDelay >= 0) {
 3             producer.setDeliveryDelay(this.deliveryDelay);
 4         }
 5         if (isExplicitQosEnabled()) {
 6             producer.send(message, getDeliveryMode(), getPriority(), getTimeToLive());
 7         }
 8         else {
 9             producer.send(message);
10         }
11     }

1     public Message receive(Destination destination) throws JmsException {
2         return receiveSelected(destination, null);
3     }

1     public Message receiveSelected(final Destination destination, @Nullable final String messageSelector) throws JmsException {
2         return execute(session -> doReceive(session, destination, messageSelector), true);
3     }

1     protected Message doReceive(Session session, Destination destination, @Nullable String messageSelector)
2             throws JMSException {
3 
4         return doReceive(session, createConsumer(session, destination, messageSelector));
5     }

 1     protected Message doReceive(Session session, MessageConsumer consumer) throws JMSException {
 2         try {
 3             // Use transaction timeout (if available).
 4             long timeout = getReceiveTimeout();
 5             ConnectionFactory connectionFactory = getConnectionFactory();
 6             JmsResourceHolder resourceHolder = null;
 7             if (connectionFactory != null) {
 8                 resourceHolder = (JmsResourceHolder) TransactionSynchronizationManager.getResource(connectionFactory);
 9             }
10             if (resourceHolder != null && resourceHolder.hasTimeout()) {
11                 timeout = Math.min(timeout, resourceHolder.getTimeToLiveInMillis());
12             }
13             Message message = receiveFromConsumer(consumer, timeout);
14             if (session.getTransacted()) {
15                 // Commit necessary - but avoid commit call within a JTA transaction.
16                 if (isSessionLocallyTransacted(session)) {
17                     // Transacted session created by this template -> commit.
18                     JmsUtils.commitIfNecessary(session);
19                 }
20             }
21             else if (isClientAcknowledge(session)) {
22                 // Manually acknowledge message, if any.
23                 if (message != null) {
24                     message.acknowledge();
25                 }
26             }
27             return message;
28         }
29         finally {
30             JmsUtils.closeMessageConsumer(consumer);
31         }
32     }

 1     protected Message receiveFromConsumer(MessageConsumer consumer, long timeout) throws JMSException {
 2         if (timeout > 0) {
 3             return consumer.receive(timeout);
 4         }
 5         else if (timeout < 0) {
 6             return consumer.receiveNoWait();
 7         }
 8         else {
 9             return consumer.receive();
10         }
11     }

1     public void afterPropertiesSet() {
2         //驗證connectionFactory
3         super.afterPropertiesSet();
4         //驗證配置文件
5         validateConfiguration();
6         初始化
7         initialize();
8     }

 1     public void initialize() throws JmsException {
 2         try {
 3             //lifecycleMonitor用於控制生命周期的同步處理
 4             synchronized (this.lifecycleMonitor) {
 5                 this.active = true;
 6                 this.lifecycleMonitor.notifyAll();
 7             }
 8             doInitialize();
 9         }
10         catch (JMSException ex) {
11             synchronized (this.sharedConnectionMonitor) {
12                 ConnectionFactoryUtils.releaseConnection(this.sharedConnection, getConnectionFactory(), this.autoStartup);
13                 this.sharedConnection = null;
14             }
15             throw convertJmsAccessException(ex);
16         }
17     }

1     protected void doInitialize() throws JMSException {
2         synchronized (this.lifecycleMonitor) {
3             for (int i = 0; i < this.concurrentConsumers; i++) {
4                 scheduleNewInvoker();
5             }
6         }
7     }

1     private void scheduleNewInvoker() {
2         AsyncMessageListenerInvoker invoker = new AsyncMessageListenerInvoker();
3         if (rescheduleTaskIfNecessary(invoker)) {
4             // This should always be true, since we're only calling this when active.
5             this.scheduledInvokers.add(invoker);
6         }
7     }

 1     protected final boolean rescheduleTaskIfNecessary(Object task) {
 2         if (this.running) {
 3             try {
 4                 doRescheduleTask(task);
 5             }
 6             catch (RuntimeException ex) {
 7                 logRejectedTask(task, ex);
 8                 this.pausedTasks.add(task);
 9             }
10             return true;
11         }
12         else if (this.active) {
13             this.pausedTasks.add(task);
14             return true;
15         }
16         else {
17             return false;
18         }
19     }

1     protected void doRescheduleTask(Object task) {
2         Assert.state(this.taskExecutor != null, "No TaskExecutor available");
3         this.taskExecutor.execute((Runnable) task);
4     }

 1         public void run() {
 2             //並發控制
 3             synchronized (lifecycleMonitor) {
 4                 activeInvokerCount++;
 5                 lifecycleMonitor.notifyAll();
 6             }
 7             boolean messageReceived = false;
 8             try {
 9                 //根據每個任務設置的最大處理消息數量而做不同的處理
10                 //小於0默認為無限制，一直能接収消息
11                 if (maxMessagesPerTask < 0) {
12                     messageReceived = executeOngoingLoop();
13                 }
14                 else {
15                     int messageCount = 0;
16                     //消息數量控制，一旦超出數量則停止循環
17                     while (isRunning() && messageCount < maxMessagesPerTask) {
18                         messageReceived = (invokeListener() || messageReceived);
19                         messageCount++;
20                     }
21                 }
22             }
23             catch (Throwable ex) {
24                 //清理操作，包括關閉session等
25                 clearResources();
26                 if (!this.lastMessageSucceeded) {
27                     // We failed more than once in a row or on startup -
28                     // wait before first recovery attempt.
29                     waitBeforeRecoveryAttempt();
30                 }
31                 this.lastMessageSucceeded = false;
32                 boolean alreadyRecovered = false;
33                 synchronized (recoveryMonitor) {
34                     if (this.lastRecoveryMarker == currentRecoveryMarker) {
35                         handleListenerSetupFailure(ex, false);
36                         recoverAfterListenerSetupFailure();
37                         currentRecoveryMarker = new Object();
38                     }
39                     else {
40                         alreadyRecovered = true;
41                     }
42                 }
43                 if (alreadyRecovered) {
44                     handleListenerSetupFailure(ex, true);
45                 }
46             }
47             finally {
48                 synchronized (lifecycleMonitor) {
49                     decreaseActiveInvokerCount();
50                     lifecycleMonitor.notifyAll();
51                 }
52                 if (!messageReceived) {
53                     this.idleTaskExecutionCount++;
54                 }
55                 else {
56                     this.idleTaskExecutionCount = 0;
57                 }
58                 synchronized (lifecycleMonitor) {
59                     if (!shouldRescheduleInvoker(this.idleTaskExecutionCount) || !rescheduleTaskIfNecessary(this)) {
60                         // We're shutting down completely.
61                         scheduledInvokers.remove(this);
62                         if (logger.isDebugEnabled()) {
63                             logger.debug("Lowered scheduled invoker count: " + scheduledInvokers.size());
64                         }
65                         lifecycleMonitor.notifyAll();
66                         clearResources();
67                     }
68                     else if (isRunning()) {
69                         int nonPausedConsumers = getScheduledConsumerCount() - getPausedTaskCount();
70                         if (nonPausedConsumers < 1) {
71                             logger.error("All scheduled consumers have been paused, probably due to tasks having been rejected. " +
72                                     "Check your thread pool configuration! Manual recovery necessary through a start() call.");
73                         }
74                         else if (nonPausedConsumers < getConcurrentConsumers()) {
75                             logger.warn("Number of scheduled consumers has dropped below concurrentConsumers limit, probably " +
76                                     "due to tasks having been rejected. Check your thread pool configuration! Automatic recovery " +
77                                     "to be triggered by remaining consumers.");
78                         }
79                     }
80                 }
81             }
82         }

 1         private boolean executeOngoingLoop() throws JMSException {
 2             boolean messageReceived = false;
 3             boolean active = true;
 4             while (active) {
 5                 synchronized (lifecycleMonitor) {
 6                     boolean interrupted = false;
 7                     boolean wasWaiting = false;
 8                     //如果當前任務已經處於激活狀態但是卻給了暫時終止的命令
 9                     while ((active = isActive()) && !isRunning()) {
10                         if (interrupted) {
11                             throw new IllegalStateException("Thread was interrupted while waiting for " +
12                                     "a restart of the listener container, but container is still stopped");
13                         }
14                         if (!wasWaiting) {
15                             //如果並非處於等待狀態則說明是第一次執行，需要將激活任務數量減少
16                             decreaseActiveInvokerCount();
17                         }
18                         //開始進入等待狀態，等待任務的恢復命令
19                         wasWaiting = true;
20                         try {
21                             //通過wait等待，也就是等待notify或者notifyAll
22                             lifecycleMonitor.wait();
23                         }
24                         catch (InterruptedException ex) {
25                             // Re-interrupt current thread, to allow other threads to react.
26                             Thread.currentThread().interrupt();
27                             interrupted = true;
28                         }
29                     }
30                     if (wasWaiting) {
31                         activeInvokerCount++;
32                     }
33                     if (scheduledInvokers.size() > maxConcurrentConsumers) {
34                         active = false;
35                     }
36                 }
37                 //正常處理流程
38                 if (active) {
39                     messageReceived = (invokeListener() || messageReceived);
40                 }
41             }
42             return messageReceived;
43         }

 1         private boolean invokeListener() throws JMSException {
 2             this.currentReceiveThread = Thread.currentThread();
 3             try {
 4                 //初始化資源包括首次創建的時候創建session和consumer
 5                 initResourcesIfNecessary();
 6                 boolean messageReceived = receiveAndExecute(this, this.session, this.consumer);
 7                 //改變標誌位，信息成功處理
 8                 this.lastMessageSucceeded = true;
 9                 return messageReceived;
10             }
11             finally {
12                 this.currentReceiveThread = null;
13             }
14         }

 1     protected boolean receiveAndExecute(
 2             Object invoker, @Nullable Session session, @Nullable MessageConsumer consumer)
 3             throws JMSException {
 4 
 5         if (this.transactionManager != null) {
 6             // Execute receive within transaction.
 7             TransactionStatus status = this.transactionManager.getTransaction(this.transactionDefinition);
 8             boolean messageReceived;
 9             try {
10                 messageReceived = doReceiveAndExecute(invoker, session, consumer, status);
11             }
12             catch (JMSException | RuntimeException | Error ex) {
13                 rollbackOnException(this.transactionManager, status, ex);
14                 throw ex;
15             }
16             this.transactionManager.commit(status);
17             return messageReceived;
18         }
19 
20         else {
21             // Execute receive outside of transaction.
22             return doReceiveAndExecute(invoker, session, consumer, null);
23         }
24     }

 1     protected boolean doReceiveAndExecute(Object invoker, @Nullable Session session,
 2             @Nullable MessageConsumer consumer, @Nullable TransactionStatus status) throws JMSException {
 3 
 4         Connection conToClose = null;
 5         Session sessionToClose = null;
 6         MessageConsumer consumerToClose = null;
 7         try {
 8             Session sessionToUse = session;
 9             boolean transactional = false;
10             if (sessionToUse == null) {
11                 sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession(
12                         obtainConnectionFactory(), this.transactionalResourceFactory, true);
13                 transactional = (sessionToUse != null);
14             }
15             if (sessionToUse == null) {
16                 Connection conToUse;
17                 if (sharedConnectionEnabled()) {
18                     conToUse = getSharedConnection();
19                 }
20                 else {
21                     conToUse = createConnection();
22                     conToClose = conToUse;
23                     conToUse.start();
24                 }
25                 sessionToUse = createSession(conToUse);
26                 sessionToClose = sessionToUse;
27             }
28             MessageConsumer consumerToUse = consumer;
29             if (consumerToUse == null) {
30                 consumerToUse = createListenerConsumer(sessionToUse);
31                 consumerToClose = consumerToUse;
32             }
33             //接收消息
34             Message message = receiveMessage(consumerToUse);
35             if (message != null) {
36                 if (logger.isDebugEnabled()) {
37                     logger.debug("Received message of type [" + message.getClass() + "] from consumer [" +
38                             consumerToUse + "] of " + (transactional ? "transactional " : "") + "session [" +
39                             sessionToUse + "]");
40                 }
41                 //模板方法，當消息接收且在未處理前給子類機會做相應的處理，
42                 messageReceived(invoker, sessionToUse);
43                 boolean exposeResource = (!transactional && isExposeListenerSession() &&
44                         !TransactionSynchronizationManager.hasResource(obtainConnectionFactory()));
45                 if (exposeResource) {
46                     TransactionSynchronizationManager.bindResource(
47                             obtainConnectionFactory(), new LocallyExposedJmsResourceHolder(sessionToUse));
48                 }
49                 try {
50                     //激活監聽器
51                     doExecuteListener(sessionToUse, message);
52                 }
53                 catch (Throwable ex) {
54                     if (status != null) {
55                         if (logger.isDebugEnabled()) {
56                             logger.debug("Rolling back transaction because of listener exception thrown: " + ex);
57                         }
58                         status.setRollbackOnly();
59                     }
60                     handleListenerException(ex);
61                     // Rethrow JMSException to indicate an infrastructure problem
62                     // that may have to trigger recovery...
63                     if (ex instanceof JMSException) {
64                         throw (JMSException) ex;
65                     }
66                 }
67                 finally {
68                     if (exposeResource) {
69                         TransactionSynchronizationManager.unbindResource(obtainConnectionFactory());
70                     }
71                 }
72                 // Indicate that a message has been received.
73                 return true;
74             }
75             else {
76                 if (logger.isTraceEnabled()) {
77                     logger.trace("Consumer [" + consumerToUse + "] of " + (transactional ? "transactional " : "") +
78                             "session [" + sessionToUse + "] did not receive a message");
79                 }
80                 //接收到空消息的處理
81                 noMessageReceived(invoker, sessionToUse);
82                 // Nevertheless call commit, in order to reset the transaction timeout (if any).
83                 if (shouldCommitAfterNoMessageReceived(sessionToUse)) {
84                     commitIfNecessary(sessionToUse, null);
85                 }
86                 // Indicate that no message has been received.
87                 return false;
88             }
89         }
90         finally {
91             JmsUtils.closeMessageConsumer(consumerToClose);
92             JmsUtils.closeSession(sessionToClose);
93             ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), true);
94         }
95     }

 1     protected void doExecuteListener(Session session, Message message) throws JMSException {
 2         if (!isAcceptMessagesWhileStopping() && !isRunning()) {
 3             if (logger.isWarnEnabled()) {
 4                 logger.warn("Rejecting received message because of the listener container " +
 5                         "having been stopped in the meantime: " + message);
 6             }
 7             rollbackIfNecessary(session);
 8             throw new MessageRejectedWhileStoppingException();
 9         }
10 
11         try {
12             invokeListener(session, message);
13         }
14         catch (JMSException | RuntimeException | Error ex) {
15             rollbackOnExceptionIfNecessary(session, ex);
16             throw ex;
17         }
18         commitIfNecessary(session, message);
19     }

 1     protected void invokeListener(Session session, Message message) throws JMSException {
 2         Object listener = getMessageListener();
 3 
 4         if (listener instanceof SessionAwareMessageListener) {
 5             doInvokeListener((SessionAwareMessageListener) listener, session, message);
 6         }
 7         else if (listener instanceof MessageListener) {
 8             doInvokeListener((MessageListener) listener, message);
 9         }
10         else if (listener != null) {
11             throw new IllegalArgumentException(
12                     "Only MessageListener and SessionAwareMessageListener supported: " + listener);
13         }
14         else {
15             throw new IllegalStateException("No message listener specified - see property 'messageListener'");
16         }
17     }

 1     protected void doInvokeListener(SessionAwareMessageListener listener, Session session, Message message)
 2             throws JMSException {
 3 
 4         Connection conToClose = null;
 5         Session sessionToClose = null;
 6         try {
 7             Session sessionToUse = session;
 8             if (!isExposeListenerSession()) {
 9                 // We need to expose a separate Session.
10                 conToClose = createConnection();
11                 sessionToClose = createSession(conToClose);
12                 sessionToUse = sessionToClose;
13             }
14             // Actually invoke the message listener...
15             listener.onMessage(message, sessionToUse);
16             // Clean up specially exposed Session, if any.
17             if (sessionToUse != session) {
18                 if (sessionToUse.getTransacted() && isSessionLocallyTransacted(sessionToUse)) {
19                     // Transacted session created by this container -> commit.
20                     JmsUtils.commitIfNecessary(sessionToUse);
21                 }
22             }
23         }
24         finally {
25             JmsUtils.closeSession(sessionToClose);
26             JmsUtils.closeConnection(conToClose);
27         }
28     }

1     protected void doInvokeListener(MessageListener listener, Message message) throws JMSException {
2         listener.onMessage(message);
3     }

 1     protected void commitIfNecessary(Session session, @Nullable Message message) throws JMSException {
 2         // Commit session or acknowledge message.
 3         if (session.getTransacted()) {
 4             // Commit necessary - but avoid commit call within a JTA transaction.
 5             if (isSessionLocallyTransacted(session)) {
 6                 // Transacted session created by this container -> commit.
 7                 JmsUtils.commitIfNecessary(session);
 8             }
 9         }
10         else if (message != null && isClientAcknowledge(session)) {
11             message.acknowledge();
12         }
13     }

1     public static void commitIfNecessary(Session session) throws JMSException {
2         Assert.notNull(session, "Session must not be null");
3         try {
4             session.commit();
5         }
6         catch (javax.jms.TransactionInProgressException | javax.jms.IllegalStateException ex) {
7             // Ignore -> can only happen in case of a JTA transaction.
8         }
9     }