消息事務

消息事務，是保證消息傳遞原子性的一個重要特征，和JDBC的事務特征類似。

一個事務性發(fā)送，其中一組消息要么能夠全部保證到達服務器，要么都不到達服務器。

生產者、消費者與消息服務器直接都支持事務性；

ActiveMQ的事務主要偏向在生產者的應用。

ActiveMQ消息事務流程圖：

1.生產者事務：

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

方式一：

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

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

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

            //創(chuàng)建生產者
            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);
            }

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


    }

結果，沒有發(fā)送出去

方式二：

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

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

}

/**
 * 消息發(fā)送的業(yè)務類
 */
@Service
public class MessageService {

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

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

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

}

2.消費者事務

/**
 * 用于監(jiān)聽消息類（既可以用于隊列的監(jiān)聽，也可以用于主題監(jiān)聽）
 */
@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會重發(fā)消息，一共重發(fā)6次
                } catch (JMSException e1) {
                    e1.printStackTrace();
                }
            }

        }
    }

}

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

以上就是動力節(jié)點小編介紹的"ActiveMQ的消息事務"，希望對大家有幫助，想了解更多可查看ActiveMQ教程。動力節(jié)點在線學習教程，針對沒有任何Java基礎的讀者學習,讓你從入門到精通,主要介紹了一些Java基礎的核心知識，讓同學們更好更方便的學習和了解Java編程，感興趣的同學可以關注一下。