消息事務(wù)

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

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

生產(chǎn)者、消費(fèi)者與消息服務(wù)器直接都支持事務(wù)性；

ActiveMQ的事務(wù)主要偏向在生產(chǎn)者的應(yīng)用。

ActiveMQ消息事務(wù)流程圖：

1.生產(chǎn)者事務(wù)：

沒有加入事務(wù)的時候，會有部分信息過去，結(jié)果如圖：

方式一：

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

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

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

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

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


    }

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

方式二：

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

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

}

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

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

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

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

}

2.消費(fèi)者事務(wù)

/**
 * 用于監(jiān)聽消息類（既可以用于隊(duì)列的監(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;

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

        }
    }

}

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

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