spring activeMQ 开启事物接收消息时自定义确认消息

我会带着你远行 2022-07-16 23:15 191阅读 0赞

# 1:环境和版本 #

java:jdk7

spring:4.1.3

activemq:5.8.0

# 2:spring与activeMQ的结合配置 #

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
    ">
    
    
        
        <bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
            <property name="brokerURL" value="tcp://localhost:61636"></property>
        </bean>
        
        <bean id="poolConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"  >
            <property name="connectionFactory" ref="activeMQConnectionFactory" />
            <property name="maxConnections" value="10"/>
        </bean>
        
        <bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
            <property name="targetConnectionFactory" ref="activeMQConnectionFactory"/>
        </bean>
        <bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
            <property name="targetConnectionFactory" ref="activeMQConnectionFactory"/>
        </bean>
    
    
        
        
        <bean id="senderJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
            
            <property name="connectionFactory" ref="singleConnectionFactory"/>
    
            
            <property name="deliveryMode" value="2"/>
    
            <property name="sessionTransacted" value="true"/>
            <property name="sessionAcknowledgeModeName" value="AUTO_ACKNOWLEDGE"/>
        </bean>
        
        <bean id="activeMQQueue" class="org.apache.activemq.command.ActiveMQQueue">
            <constructor-arg value="FirstQueue"/>
        </bean>
        
        <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
            <constructor-arg value="topic"/>
        </bean>
    
        
        
        <bean id="receiverJmsTemplate" class="com.system.freemwork.amq.SimpleJmsTemplate">
            
            <property name="connectionFactory" ref="singleConnectionFactory"/>
    
            
            <property name="sessionTransacted" value="false"/>
            <property name="sessionAcknowledgeModeName" value="CLIENT_ACKNOWLEDGE"/>
            <property name="receiveTimeout" value="1000"/>
            <property name="autoAcknowledge" value="true"/>
        </bean>
    </beans>

# 3:编写send测试类 #

package com.test.spring;
    
    import org.junit.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsTemplate;
    
    /**
     * 类描述:sender测试类
     *
     * @author fengyong
     * @version 1.0
     * @since 1.0
     * Created by fengyong on 16/8/3 下午7:45.
     */
    public class ActiveMqSender extends BaseTest {
        @Autowired
        private JmsTemplate senderJmsTemplate;
    
        @Test
        public void activeMq(){
            for(int i = 1;i<=10;i++){
                senderJmsTemplate.convertAndSend("FirstQueue","我是第"+i+"个");
            }
            System.out.print("全部执行完毕!!!");
        }
    }

# 4:编写receiver测试类 #

package com.test.spring;
    
    import com.system.freemwork.amq.SimpleJmsTemplate;
    import org.apache.activemq.command.ActiveMQQueue;
    import org.junit.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsTemplate;
    
    import javax.jms.JMSException;
    import javax.jms.TextMessage;
    
    /**
     * 类描述:receiver测试类
     *
     * @author fengyong
     * @version 1.0
     * @since 1.0
     * Created by fengyong on 16/8/8 下午5:28.
     */
    public class ActiveMqReceiver extends BaseTest{
        @Autowired
        private SimpleJmsTemplate receiverJmsTemplate;
        @Autowired
        private ActiveMQQueue activeMQQueue;
    
        /**
         * 坑爹的方法,如果session事物设置为true,receiver直接将sessioin进行commit,
         *
         * 如果设置为false,receiver方法会直接判断进行消息确认,无法做到手动的消息确认,所以一旦发生异常,这条消息不会回到消息队列中
         *
         * session的提交可以认为是消息确认收到
         * @throws JMSException
         */
        @Test
        public void receiver() throws JMSException {
            int i=1;
            while (true){
                i++;
                TextMessage message = (TextMessage)receiverJmsTemplate.receive(activeMQQueue);
                if (null != message) {
                    System.out.println("收到消息==================" + message.getText());
    
                } else {
                    System.out.print("超时10秒");
                    break;
                }
            }
        }
    }

#### 注:如果session事物设置为true,receiver直接将sessioin进行commit.源码如下 ####

if (session.getTransacted()) {
       // Commit necessary - but avoid commit call within a JTA transaction.
       if (isSessionLocallyTransacted(session)) {
          // Transacted session created by this template -> commit.
          JmsUtils.commitIfNecessary(session);
       }
    }

#### 如果session事物设置为false,receiver方法会直接判断进行消息确认,无法做到手动的消息确认,所以一旦发生异常,这条消息不会回到消息队列中.源码如下 ####

else if (isClientAcknowledge(session)) {
       // Manually acknowledge message, if any.
       if (message != null) {
          message.acknowledge();
       }
    }

### 所以需要修改源码不让其在receiver的时候自动确认收到消息 ###

# 5:新建SimpleJmsTemplate继承JmsTemplate #

package com.system.freemwork.amq;
    
    import org.springframework.jms.JmsException;
    import org.springframework.jms.connection.ConnectionFactoryUtils;
    import org.springframework.jms.connection.JmsResourceHolder;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.SessionCallback;
    import org.springframework.jms.support.JmsUtils;
    import org.springframework.transaction.support.TransactionSynchronizationManager;
    import org.springframework.util.Assert;
    import sun.misc.resources.Messages_ja;
    
    import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    
    /**
     * 类描述:自定义JmsTemplate,实现客户手动确认
     *
     * @author fengyong
     * @version 1.0
     * @since 1.0
     * Created by fengyong on 16/8/10 上午10:03.
     */
    public class SimpleJmsTemplate extends JmsTemplate {
    
    
        private final JmsTemplateResourceFactory transactionalResourceFactory = new JmsTemplateResourceFactory();
        /**
         * 是否开启手动确认标记
         */
        private Boolean autoAcknowledge;
    
        MessageConsumer consumer = null;
        Session sessionToClose = null;
        Connection conToClose = null;
        boolean startConnection = false;
    
        /**
         * 接收消息
         * @param session
         * @param consumer
         * @return
         * @throws JMSException
         */
        protected Message doReceive(Session session, MessageConsumer consumer) throws JMSException {
            try {
                this.consumer = consumer;
                // Use transaction timeout (if available).
                long timeout = getReceiveTimeout();
                JmsResourceHolder resourceHolder =
                        (JmsResourceHolder) TransactionSynchronizationManager.getResource(getConnectionFactory());
                if (resourceHolder != null && resourceHolder.hasTimeout()) {
                    timeout = Math.min(timeout, resourceHolder.getTimeToLiveInMillis());
                }
                Message message = doReceive(consumer, timeout);
                if (session.getTransacted()) {
                    // Commit necessary - but avoid commit call within a JTA transaction.
                    // 如果开启了jta事物,那么不会进行提交,jta事物会直接覆盖掉session事物
                    if (isSessionLocallyTransacted(session)) {
                        // Transacted session created by this template -> commit.
                        JmsUtils.commitIfNecessary(session);
                    }
                }
                //autoAcknowledge如果为真,不进行自动确认
                else if (isClientAcknowledge(session) && !autoAcknowledge) {
                    // Manually acknowledge message, if any.
                    if (message != null) {
                        message.acknowledge();
                    }
                }
                return message;
            }
            finally {
                consumer = null;
            }
        }
    
        /**
         * 自定义的消息确认,关闭consumer和sesseionToClose是父类本身就要执行的,这里直接拷贝下来,能不改的地方尽量不改
         * 该子类只是为了自定义确认消息
         * @param message
         * @throws JMSException
         */
        public void  msgAckAndcloseSession(Message message) throws JMSException {
            message.acknowledge();
            JmsUtils.closeMessageConsumer(consumer);
            JmsUtils.closeSession(sessionToClose);
            ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), startConnection);
        }
    
        /**
         * 由于上面的doReceive(Session session, MessageConsumer consumer)需要调用这个方法,
         * 而在父类里面这个方法是私有的,所以直接拷贝下来了
         * @param consumer
         * @param timeout
         * @return
         * @throws JMSException
         */
        private Message doReceive(MessageConsumer consumer, long timeout) throws JMSException {
            if (timeout == RECEIVE_TIMEOUT_NO_WAIT) {
                return consumer.receiveNoWait();
            }
            else if (timeout > 0) {
                return consumer.receive(timeout);
            }
            else {
                return consumer.receive();
            }
        }
    
    
        /**
         * 该方法是为了防止确认消息前session被关闭,不然确认消息前session关闭会导致异常发生
         * transactionalResourceFactory在父类中是私有且不可修改,因为只有这一个方法用到了transactionalResourceFactory
         * 所以直接将JmsTemplateResourceFactory拷贝下来使用
         * @param action
         * @param startConnection
         * @param <T>
         * @return
         * @throws JmsException
         */
        public <T> T execute(SessionCallback<T> action, boolean startConnection) throws JmsException {
            Assert.notNull(action, "Callback object must not be null");
            this.startConnection = startConnection;
    
            try {
                Session sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession(
                        getConnectionFactory(), transactionalResourceFactory, startConnection);
                if (sessionToUse == null) {
                    conToClose = createConnection();
                    sessionToClose = createSession(conToClose);
                    if (startConnection) {
                        conToClose.start();
                    }
                    sessionToUse = sessionToClose;
                }
                if (logger.isDebugEnabled()) {
                    logger.debug("Executing callback on JMS Session: " + sessionToUse);
                }
                return action.doInJms(sessionToUse);
            }
            catch (JMSException ex) {
                throw convertJmsAccessException(ex);
            }
            finally {
                sessionToClose = null;
                conToClose = null;
                startConnection = false;
            }
        }
    
    
        /**
         * Sets new 是否开启手动确认标记.
         *
         * @param autoAcknowledge New value of 是否开启手动确认标记.
         */
        public void setAutoAcknowledge(Boolean autoAcknowledge) {
            this.autoAcknowledge = autoAcknowledge;
        }
    
        /**
         * Gets 是否开启手动确认标记.
         *
         * @return Value of 是否开启手动确认标记.
         */
        public Boolean getAutoAcknowledge() {
            return autoAcknowledge;
        }
    
    
        /**
         * 直接拷贝下来的
         */
        private class JmsTemplateResourceFactory implements ConnectionFactoryUtils.ResourceFactory {
    
            @Override
            public Connection getConnection(JmsResourceHolder holder) {
                return SimpleJmsTemplate.this.getConnection(holder);
            }
    
            @Override
            public Session getSession(JmsResourceHolder holder) {
                return SimpleJmsTemplate.this.getSession(holder);
            }
    
            @Override
            public Connection createConnection() throws JMSException {
                return SimpleJmsTemplate.this.createConnection();
            }
    
            @Override
            public Session createSession(Connection con) throws JMSException {
                return SimpleJmsTemplate.this.createSession(con);
            }
    
            @Override
            public boolean isSynchedLocalTransactionAllowed() {
                return SimpleJmsTemplate.this.isSessionTransacted();
            }
        }
    
    }

# 6:第二个receiver测试类 #

package com.test.spring;
    
    import com.system.freemwork.amq.SimpleJmsTemplate;
    import org.apache.activemq.command.ActiveMQQueue;
    import org.junit.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsTemplate;
    
    import javax.jms.JMSException;
    import javax.jms.TextMessage;
    
    /**
     * 类描述:receiver测试类
     *
     * @author fengyong
     * @version 1.0
     * @since 1.0
     * Created by fengyong on 16/8/8 下午5:28.
     */
    public class ActiveMqReceiver extends BaseTest{
        @Autowired
        private SimpleJmsTemplate receiverJmsTemplate;
        @Autowired
        private ActiveMQQueue activeMQQueue;
    
       
        @Test
        public void recerver() throws JMSException {
            int i=1;
            while (true){
                i++;
                TextMessage message = (TextMessage)receiverJmsTemplate.receive(activeMQQueue);
                if (null != message) {
                    System.out.println("收到消息==================" + message.getText());
                    if(i==4){
                        throw new RuntimeException("Exception");
                    }
                    receiverJmsTemplate.msgAckAndcloseSession(message);
                } else {
                    System.out.print("超时10秒");
                    break;
                }
            }
        }
    }

# 7:测试结果 #

收到消息==================我是第1个
    收到消息==================我是第2个
    收到消息==================我是第3个
    
    java.lang.RuntimeException
    : Exception
    	at com.test.spring.ActiveMqReceiver.show(ActiveMqReceiver.java:43)
    	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    	at java.lang.reflect.Method.invoke(Method.java:606)
    	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
    	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
    	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
    	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
    	at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:73)

#### 注:接收第三个时失败,虽然这里打印出来了,但是第三个消息并没有被确认接收,同步自定义接收消息修改成功 ####

####     ####