1. 本地事務(wù) ????

1.1. 在一個(gè)JMS客戶端, 可以使用本地事務(wù)來組合消息的發(fā)送和接收。JMS Session接口提供了commit和rollback方法。事務(wù)提交意味著生產(chǎn)的所有消息被發(fā)送, 消費(fèi)的所有消息被確認(rèn); 事務(wù)回滾意味著生產(chǎn)的所有消息被銷毀, 消費(fèi)的所有消息被恢復(fù)并重新提交, 除非它們已經(jīng)過期。 ????

1.2. 事務(wù)性的會(huì)話總是牽涉到事務(wù)處理中, commit或rollback方法一旦被調(diào)用, 一個(gè)事務(wù)就結(jié)束了, 而另一個(gè)事務(wù)被開始。關(guān)閉事務(wù)性會(huì)話將回滾其中的事務(wù)。 ????

1.3. 需要注意的還有一個(gè), 消息的生產(chǎn)和消費(fèi)不能包含在同一個(gè)事務(wù)中。

2. 創(chuàng)建事務(wù)會(huì)話

2.1. 創(chuàng)建事務(wù)會(huì)話的時(shí)候transacted參數(shù)要設(shè)置為true, acknowledgeMode參數(shù)要設(shè)置為Session.SESSION_TRANSACTED。

// 1. 創(chuàng)建一個(gè)連接工廠 ConnectionFactory cf = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL); // 2. 創(chuàng)建連接 Connection conn = cf.createConnection(); // 4. 創(chuàng)建會(huì)話 Session session = conn.createSession(true, Session.SESSION_TRANSACTED);

2.2. Session.SESSION_TRANSACTED值是專門為事務(wù)會(huì)話使用的。

3. 例子

3.1. 創(chuàng)建一個(gè)名為JMSTransacted的Java項(xiàng)目, 同時(shí)拷入相關(guān)jar包

3.2. 編輯MyProducer.java

package com.jmsapp.tra;import javax.jms.DeliveryMode; import javax.jms.JMSException; import javax.jms.MapMessage; import javax.jms.Queue; import javax.jms.QueueConnection; import javax.jms.QueueConnectionFactory; import javax.jms.QueueSender; import javax.jms.QueueSession; import javax.jms.Session; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory;public class MyProducer {// 默認(rèn)連接用戶名private static final String dftUsr = ActiveMQConnection.DEFAULT_USER;// 默認(rèn)用戶密碼private static final String dftPwd = ActiveMQConnection.DEFAULT_PASSWORD;// 默認(rèn)連接地址private static final String dftUrl = ActiveMQConnection.DEFAULT_BROKER_URL;// 隊(duì)列名稱private static final String queueName = "queueMapMsgTransacted";public static void main(String[] args) {// 1. 創(chuàng)建一個(gè)連接工廠QueueConnectionFactory cf = new ActiveMQConnectionFactory(dftUsr, dftPwd, dftUrl);// 連接對(duì)象QueueConnection conn = null;// 會(huì)話對(duì)象QueueSession session = null;try {// 2. 創(chuàng)建連接conn = cf.createQueueConnection();// 3. 啟動(dòng)連接conn.start();// 4. 創(chuàng)建會(huì)話session = conn.createQueueSession(true, Session.SESSION_TRANSACTED);// 5. 創(chuàng)建消息目的地。如果是點(diǎn)對(duì)點(diǎn), 那么它的實(shí)現(xiàn)是Queue; 如果是訂閱模式, 那它的實(shí)現(xiàn)是Topic。這里我們創(chuàng)建一個(gè)名為queueMapMsgTransacted的消息隊(duì)列。Queue queue = session.createQueue(queueName);// 6. 消息生產(chǎn)者QueueSender sender = session.createSender(null);// 7. 創(chuàng)建文本消息和發(fā)送消息MapMessage user = session.createMapMessage();user.setLong("id", 100000000000L);user.setString("name", "lisi");sender.send(queue, user, DeliveryMode.PERSISTENT, 9, 1000 * 60 * 60);MapMessage userExtend = session.createMapMessage();userExtend.setChar("sex", '男');userExtend.setBoolean("married", false);sender.send(queue, userExtend, DeliveryMode.PERSISTENT, 5, 1000 * 60 * 60);// 8. 提交事務(wù)session.commit();} catch (JMSException e) {if(null != session) {try {session.rollback();} catch (JMSException e1) {e1.printStackTrace();}}e.printStackTrace();} finally {try {if (session != null) {session.close();}} catch (JMSException e1) {e1.printStackTrace();} finally {if (conn != null) {try {conn.close();} catch (JMSException e) {e.printStackTrace();}}}}} }

3.3. 點(diǎn)對(duì)點(diǎn)消息過期, 會(huì)把它存放到一個(gè)ActiveMQ.DLQ的隊(duì)列中

3.4. 編輯MyConsumer.java

package com.jmsapp.tra;import java.util.Enumeration; import javax.jms.JMSException; import javax.jms.MapMessage; import javax.jms.Queue; import javax.jms.QueueConnection; import javax.jms.QueueConnectionFactory; import javax.jms.QueueReceiver; import javax.jms.QueueSession; import javax.jms.Session; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory;public class MyConsumer {// 默認(rèn)連接用戶名private static final String dftUsr = ActiveMQConnection.DEFAULT_USER;// 默認(rèn)用戶密碼private static final String dftPwd = ActiveMQConnection.DEFAULT_PASSWORD;// 默認(rèn)連接地址private static final String dftUrl = ActiveMQConnection.DEFAULT_BROKER_URL;// 隊(duì)列名稱private static final String queueName = "queueMapMsgTransacted";public static void main(String[] args) {// 1. 創(chuàng)建一個(gè)連接工廠QueueConnectionFactory cf = new ActiveMQConnectionFactory(dftUsr, dftPwd, dftUrl);// 連接對(duì)象QueueConnection conn = null;// 會(huì)話對(duì)象QueueSession session = null;try {// 2. 創(chuàng)建連接conn = cf.createQueueConnection();// 3. 啟動(dòng)連接conn.start();// 4. 創(chuàng)建會(huì)話session = conn.createQueueSession(true, Session.SESSION_TRANSACTED);// 5. 創(chuàng)建消息目的地。如果是點(diǎn)對(duì)點(diǎn), 那么它的實(shí)現(xiàn)是Queue; 如果是訂閱模式, 那它的實(shí)現(xiàn)是Topic。這里我們創(chuàng)建一個(gè)名為queueMapMsgTransacted的消息隊(duì)列。Queue queue = session.createQueue(queueName);// 6. 消息接收者QueueReceiver receiver = session.createReceiver(queue);// 7. 創(chuàng)建文本消息和發(fā)送消息while(true) {MapMessage msg = (MapMessage) receiver.receive(3000);if(msg == null) {break;}@SuppressWarnings("unchecked")Enumeration<String> e = msg.getMapNames();while(e.hasMoreElements()) {String name = e.nextElement();System.out.println(name + " = " + msg.getObject(name));}}// 8. 提交事務(wù)session.commit();} catch (JMSException e) {if(null != session) {try {session.rollback();} catch (JMSException e1) {e1.printStackTrace();}}e.printStackTrace();} finally {try {if (session != null) {session.close();}} catch (JMSException e1) {e1.printStackTrace();} finally {if (conn != null) {try {conn.close();} catch (JMSException e) {e.printStackTrace();}}}}} }

3.5. 再次運(yùn)行MyProducer.java, 然后運(yùn)行MyConsumer.java

總結(jié)

以上是生活随笔為你收集整理的009_JMS中的事务的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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