1、ActiveMQ的本地事務(wù)

?

? ? ? ??在一個(gè)JMS客戶端，可以使用本地事務(wù)來(lái)組合消息的發(fā)送和接收。JMS Session接口提供了commit和rollback方法。事務(wù)提交意味著生產(chǎn)的所有消息被發(fā)送，消費(fèi)的所有消息被確認(rèn)；事務(wù)回滾意味著生產(chǎn)的所有消息被銷毀，消費(fèi)的所有消息被恢復(fù)并重新提交，除非它們已經(jīng)過期。 事務(wù)性的會(huì)話總是牽涉到事務(wù)處理中，commit或rollback方法一旦被調(diào)用，一個(gè)事務(wù)就結(jié)束了，而另一個(gè)事務(wù)被開始。關(guān)閉事務(wù)性會(huì)話將回滾其中的事務(wù)。 需要注意的是，如果使用請(qǐng)求/回復(fù)機(jī)制，即發(fā)送一個(gè)消息，同時(shí)希望在同一個(gè)事務(wù)中等待接收該消息的回復(fù)，那么程序?qū)⒈粧炱?#xff0c;因?yàn)橹钡绞聞?wù)提交，發(fā)送操作才會(huì)真正執(zhí)行。 需要注意的還有一個(gè)，消息的生產(chǎn)和消費(fèi)不能包含在同一個(gè)事務(wù)中。

? ? ? ?在事務(wù)狀態(tài)下進(jìn)行發(fā)送操作，消息并未真正投遞到中間件，而只有進(jìn)行session.commit操作之后，消息才會(huì)發(fā)送到中間件，再轉(zhuǎn)發(fā)到適當(dāng)?shù)南M(fèi)者進(jìn)行處理。如果是調(diào)用rollback操作，則表明，當(dāng)前事務(wù)期間內(nèi)所發(fā)送的消息都取消掉。

?

2、關(guān)于ActiveMQ本地事務(wù)的用法

public class Sender { public static void main(String[] args) throws Exception { // 1、建立ConnectionFactory工廠對(duì)象，需要填入用戶名，密碼，以及連接的地址 // 僅使用默認(rèn)。端口號(hào)為"tcp://localhost:61616" ConnectionFactory connectionFactory = new ActiveMQConnectionFactory( "zhangsan",// ActiveMQConnectionFactory.DEFAULT_USER, "123",// ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616"); // 2、通過ConnectionFactory工廠對(duì)象創(chuàng)建一個(gè)Connection連接 // 并且調(diào)用Connection的start方法開啟連接,Connection默認(rèn)是不開啟的 Connection connection = connectionFactory.createConnection(); connection.start(); // 3、通過Connection對(duì)象創(chuàng)建Session會(huì)話（上下文環(huán)境對(duì)象）， // 參數(shù)一，表示是否開啟事務(wù) // 參數(shù)二，表示的是簽收模式，一般使用的有自動(dòng)簽收和客戶端自己確認(rèn)簽收 // 第一個(gè)參數(shù)設(shè)置為true，表示開啟事務(wù) // 開啟事務(wù)后，記得要手動(dòng)提交事務(wù) Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 4、通過Session創(chuàng)建Destination對(duì)象，指的是一個(gè)客戶端用來(lái)指定生產(chǎn)消息目標(biāo)和消費(fèi)消息來(lái)源的對(duì)象。 // 在PTP模式中,Destination指的是Queue // 在發(fā)布訂閱模式中，Destination指的是Topic Destination destination = session.createQueue("queue1"); // 5、使用Session來(lái)創(chuàng)建消息對(duì)象的生產(chǎn)者或者消費(fèi)者 MessageProducer messageProducer = session.createProducer(destination); // 6、如果是，生產(chǎn)者，使用MessageProducer的setDeliverMode方法設(shè)置，消息的持久化和非持久化 messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 7、最后使用JMS規(guī)范的TextMessage形式創(chuàng)建數(shù)據(jù)(通過Session對(duì)象) // 并利用MessageProducer的send方法發(fā)送數(shù)據(jù) for (int i = 0; i < 5; i++) { TextMessage textMessage = session.createTextMessage(); textMessage.setText("我是消息" + i); messageProducer.send(textMessage); } // 手動(dòng)提交開啟的事務(wù) session.commit(); // 釋放連接 if (connection != null) { connection.close(); } } }

?

2.1 ? 開啟事務(wù)

// 3、通過Connection對(duì)象創(chuàng)建Session會(huì)話（上下文環(huán)境對(duì)象）， // 參數(shù)一，表示是否開啟事務(wù) // 參數(shù)二，表示的是簽收模式，一般使用的有自動(dòng)簽收和客戶端自己確認(rèn)簽收 // 第一個(gè)參數(shù)設(shè)置為true，表示開啟事務(wù) // 開啟事務(wù)后，記得要手動(dòng)提交事務(wù) Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

?

?2.2 ? ?提交事務(wù)

?

// 7、最后使用JMS規(guī)范的TextMessage形式創(chuàng)建數(shù)據(jù)(通過Session對(duì)象) // 并利用MessageProducer的send方法發(fā)送數(shù)據(jù) for (int i = 0; i < 5; i++) { TextMessage textMessage = session.createTextMessage(); textMessage.setText("我是消息" + i); messageProducer.send(textMessage); } // 手動(dòng)提交開啟的事務(wù) session.commit(); 創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎(jiǎng)勵(lì)來(lái)咯，堅(jiān)持創(chuàng)作打卡瓜分現(xiàn)金大獎(jiǎng)

總結(jié)

以上是生活随笔為你收集整理的JMS学习五（ActiveMQ的本地事务）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。