Activemq的消息事務

消息事務
消息事務，是保證消息傳遞原子性的一個重要特征，和JDBC的事務特征類似。
一個事務性發送，其中一組消息要么能夠全部保證到達服務器，要么都不到達服務器。
生產者、消費者與消息服務器直接都支持事務性；
ActionMQ的事務主要偏向在生產者的應用。

ActionMQ 消息事務流程圖：

一、生產者事務：

沒有加入事務的時候，會有部分信息過去，結果如圖：

方式一：

 /**
     * 事務性發送--方案一
     */
    @Test
    public void sendMessageTx(){
        //獲取連接工廠
        ConnectionFactory connectionFactory = jmsMessagingTemplate.getConnectionFactory();

        Session session = null;
        try {
            //創建連接
            Connection connection = connectionFactory.createConnection();

            /**
             * 參數一：是否開啟消息事務
             */
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

            //創建生產者
            MessageProducer producer = session.createProducer(session.createQueue(name));

            for(int i=1;i<=10;i++){
                //模擬異常
                if(i==4){
                    int a = 10/0;
                }

                TextMessage textMessage = session.createTextMessage("消息--" + i);
                producer.send(textMessage);
            }

            //注意：一旦開啟事務發送，那么就必須使用commit方法進行事務提交，否則消息無法到達MQ服務器
            session.commit();
        } catch (JMSException e) {
            e.printStackTrace();
            //消息事務回滾
            try {
                session.rollback();
            } catch (JMSException e1) {
                e1.printStackTrace();
            }
        }


    }

結果，沒有發送出去

方式二：

/**
 * ActiveMQ配置類
 */
@Configuration
public class ActiveMQConfig {

    /**
     * 添加Jms事務管理器
     */
    @Bean
    public PlatformTransactionManager createTransactionManager(ConnectionFactory connectionFactory){
        return new JmsTransactionManager(connectionFactory);
    }

}

/**
 * 消息發送的業務類
 */
@Service
public class MessageService {

    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;
    @Value("${activemq.name}")
    private String name;

    @Transactional // 對消息發送加入事務管理（同時也對JDBC數據庫的事務生效）
    public void sendMessage(){
        for(int i=1;i<=10;i++) {
            //模擬異常
            if(i==4){
                int a = 10/0;
            }

            jmsMessagingTemplate.convertAndSend(name, "消息---"+i);
        }
    }

}

二、消費者事務

/**
 * 用於監聽消息類（既可以用於隊列的監聽，也可以用於主題監聽）
 */
@Component // 放入IOC容器
public class MsgListener {

    /**
     * 接收TextMessage的方法
     */
    @JmsListener(destination = "${activemq.name}")
    public void receiveMessage(Message message,Session session){
        if(message instanceof TextMessage){
            TextMessage textMessage = (TextMessage)message;

            try {
                System.out.println("接收消息："+textMessage.getText());


                int i=10/0;

                //提交事務
                session.commit();
            } catch (JMSException e) {
                e.printStackTrace();
                //回滾事務
                try {
                    session.rollback();//一旦事務回滾，MQ會重發消息，一共重發6次
                } catch (JMSException e1) {
                    e1.printStackTrace();
                }
            }

        }
    }

}

注意如果在消費者異常了，會收到消息，然后重發6次，要是期間還是異常，就會到私信隊列中