JavaEE5学习笔记03-JMS介绍与使用

1、JavaEE5学习笔记03-JMS介绍与使用刘岩Email：1. 介绍JMS（Java Message Service：Java消息服务）也是JavaEE很重要的标准之一，企业应用组件互相需要通讯，就引出了面向消息的模式来进行消息互通。基于消息服务的方式不需要消息发送方与消息接收方进行任何的耦合，甚至可以不需要知道彼此的存在。这种通讯方式既可以是同步的、也可以是异步的（因业务需求不同开发者可自己定制）。官方的解释如下：JMS是一种与厂商无关的 API，用来访问消息收发系统。它类似于 JDBC(Java Database Connectivity)：这里，JD

2、BC 是可以用来访问许多不同关系数据库的 API，而 JMS 则提供同样与厂商无关的访问方法，以访问消息收发服务。许多厂商目前都支持 JMS，包括 IBM 的 MQSeries、BEA的 Weblogic JMS service和 Progress 的 SonicMQ，这只是几个例子。 JMS 使您能够通过消息收发服务（有时称为消息中介程序或路由器）从一个 JMS 客户机向另一个 JML 客户机发送消息。消息是 JMS 中的一种类型对象，由两部分组成：报头和消息主体。报头由路由信息以及有关该消息的元数据组成。消息主体则携带着应用程序的数据或有效负载。根据有效负载的类型来划分，可以将消息分为几种

3、类型，它们分别携带：简单文本 (TextMessage)、可序列化的对象 (ObjectMessage)、属性集合 (MapMessage)、字节流 (BytesMessage)、原始值流 (StreamMessage)，还有无有效负载的消息 (Message)。实际上读者可以把JMS当做“QQ”，每个QQ使用人可以当做企业中的一个小组件，其中一个人A君具备查询大量数据的功能，另一个人B君具备破解密码的功能。能解码的人需要能查询大量数据的人提供查询数据的服务。这个时候B君就用QQ和A君通讯：“嗨，哥们，给我一个密码的密文，我来破解！”，消息发到QQ服务器，通过网络传输将此消息传达给A君，A君收

4、到消息后开始工作，将查询结果return给服务器，服务器再将return的结果传给了B君，B君拿到密文结果后接下来自己再工作。以上是一个同步消息的例子，没有反馈的密文，B君无法进行下一步的工作，他会一直等待A君给他的反馈。如果B君下一步的工作不是严格依赖于A君的反馈信息，则完全可以采用异步的方式进行消息的传输。比如B君可以这样发消息说：“我要自己破解手中已经收到的一个密文了啊，哥们，如果你在线，给我一个回复就行！一会儿可能需要您的配合”。这样即使A君不在线，没收到这个消息，也不会影响B君下一步的工作执行。2. 角色说明JMS分为3个角色1. JMS消息生产者：顾名思义，一个消息的发起者2. J

5、MS消息的消费者：顾名思义，这个消息的实际读取、处理者3. JMS消息服务器：路由来自各方的JMS消息的服务器，通常也就是指着应用服务器，比如Weblogic、JBoss等都提供了JMS消息服务支持。另外，JMS生产者都是将消息发送到一个目的地、消息消费者也是从该目的地去取得消息。这个目的地也就是在应用服务器上。JMS消息也是通过应用服务器这个“中介公司”传达的。由此看来，JMS的消息规范体系和QQ的信息原理更加吻合了。3. 发送与消费方式点对点PTP方式：咱们还是拿QQ来举例子，这种方式就好比上面的例子，一个人和另一个人进行单一的通讯，除了服务器负责转发、路由消息，其他消息消费者不会接收该消

6、息。只有消息目的消费者才能收到此消息。发布/订阅Pub-Sub方式：这个就好比一个人往QQ群发送一个消息，所有在这个群中、并且当时在线的人都会接收到来自群里的消息。当然了这里所说的订阅就是表示在群中的人。实际上在JMS1.1规范推出后，给使用者的感觉，起码在编程上PTP方式和Pub-Sub方式差不太多，对于使用者除了在应用服务器中配置的消息目的地有些差异之外，其他并没太感觉有太大差异（底层会做出区分，根据不同类型的消息会自动调用不同类型的API，典型的多态）。统一的API接口如下：JMS1.1接口连接工厂接口ConnectionFactoryJMS连接接口Connection会话接口Sessi

7、on消息目的接口Destination消息生产者接口MessageProducer消息消费者接口MessageConsumer4. 示例代码1.PTP消息首先，我们现在JBoss当中配置一个PTP类型的消息目的可以看到我配置了一个目的叫做jbossJMS的Resource，那我们就可以往这个消息目的发送JMS消息，并且可以从这个JMS目的消费消息了。咱们先看消费者代码，当然也应该是先执行消费者代码，等消息生产者生产消息的时候，消费者才能见听到。package jms.ptp;import java.util.Properties;import javax.jms.Connection;impo

8、rt javax.jms.ConnectionFactory;import javax.jms.Destination;import javax.jms.MessageConsumer;import javax.jms.Session;import javax.jms.TextMessage;import javax.naming.Context;import javax.naming.InitialContext;import javax.naming.NamingException;/* * 同步接收P2P消息 * * author liuyan * */public class Jbos

9、sSyncMessageReciveriver /* * 消息消费者 */public void reciveMessage() try String Connection_Factory = ConnectionFactory;Context context = getInitialContext();ConnectionFactory connectionFactory = (ConnectionFactory) context.lookup(Connection_Factory);Destination dest = (Destination) context.lookup(jbossJ

10、MS);Connection connection = connectionFactory.createConnection();connection.start();Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);MessageConsumer reciver = session.createConsumer(dest);TextMessage textMessage = (TextMessage) reciver.receive();String messageText = textMes

11、sage.getText();System.out.println(textMessage： + textMessage);System.out.println(消息内容 + messageText);session.close();connection.close(); catch (Exception e) e.printStackTrace();/* * 获得应用服务上下文信息 * * return */private Context getInitialContext() String init_factory = erfaces.NamingContextFac

12、tory;String serverURL = jnp:/:1099;Context context = null;Properties properties = new Properties();properties.put(Context.INITIAL_CONTEXT_FACTORY, init_factory);properties.put(Context.PROVIDER_URL, serverURL);try context = new InitialContext(properties); catch (NamingException e) / TODO Aut

13、o-generated catch blocke.printStackTrace();return context;/* * param args */public static void main(String args) JbossSyncMessageReciveriver jbossSyncMessageReciveriver = new JbossSyncMessageReciveriver();jbossSyncMessageReciveriver.reciveMessage();以上是一个同步接收消息的代码，流程是1)：获得应用上下文2)：根据上下文获得连接工厂，此处使用的是JB

14、oss默认的连接工厂3)：从连接工厂里面获取JMS连接Connection4)：用Connection.start()方法，传输JMS消息5)：由JMS连接获得JMS消息回话Session6)：根据消息目的，由JMS会话创建消息消费者7)：同步接收消息，若没接受到消息，线程会阻塞8)：关闭JMS消息资源，session、connection下面我们来看消息生产者的代码:package jms.ptp;import java.util.Properties;import javax.jms.Connection;import javax.jms.ConnectionFactory;import

15、javax.jms.DeliveryMode;import javax.jms.Destination;import javax.jms.MessageProducer;import javax.jms.Session;import javax.jms.TextMessage;import javax.naming.Context;import javax.naming.InitialContext;import javax.naming.NamingException;/* * 消息发送者 * * author liuyan * */public class JbossMessageSend

16、er /* * 发送消息 * * throws Exception */public void sendMessage() try String Connection_Factory = ConnectionFactory;Context context = getInitialContext();ConnectionFactory connectionFactory = (ConnectionFactory) context.lookup(Connection_Factory);Destination dest = (Destination) context.lookup(jbossJMS)

17、;Connection connection = connectionFactory.createConnection();Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);MessageProducer sender = session.createProducer(dest);sender.setDeliveryMode(DeliveryMode.PERSISTENT);sender.setTimeToLive(2000);TextMessage textMessage = session.

18、createTextMessage();textMessage.setStringProperty(ConType, txt);textMessage.setText(hello 速速);sender.send(textMessage);/ 等待50秒Thread.sleep(5000);System.out.println(停顿5秒钟);textMessage.setText(welcome to JMS);sender.send(textMessage);session.close();connection.close(); catch (Exception e) e.printStack

19、Trace();private Context getInitialContext() /重复代码/* * param args * throws Exception */public static void main(String args) throws Exception JbossMessageSender jbossMessageSender = new JbossMessageSender();jbossMessageSender.sendMessage();以上是一个消息生产者的代码，流程架子和消息消费者差不多是1)：获得应用上下文2)：根据上下文获得连接工厂，此处使用的是JBo

20、ss默认的连接工厂3)：从连接工厂里面获取JMS连接Connection 4)：由JMS连接获得JMS消息回话Session5)：根据消息目的，由JMS会话创建消息生产者6)：设置消息头的一些属性7)：填充消息体8)：发送消息9)：关闭JMS消息资源，session、connection这样先执行消息消费者，苦苦等待（世间最痛苦的莫过于没有任何音讯的苦苦等待）来自目的的消息。之后再执行消息生产者看到效果如下：消息消费者的控制台：textMessage：delegator-JBossMessage:PERSISTENT, deliveryId=0消息内容hello 速速可以看出生产者生产了2个消

21、息，消费者只享用了一个消息后就继续往下执行代码，执行到close()时，退出消息目的的监听了，看来这个消息消费者很知足啊：你目的没有消息，只要我活着我就永远等着，知道你目的有一个我等待的消息出现，我才往下执行代码。下面我们不改变消息生产者，修改一下消息的消费者为异步的方式看看效果，代码如下：package jms.ptp;import java.util.Properties;import javax.jms.Connection;import javax.jms.ConnectionFactory;import javax.jms.Destination;import javax.jms.J

22、MSException;import javax.jms.Message;import javax.jms.MessageConsumer;import javax.jms.MessageListener;import javax.jms.Session;import javax.jms.TextMessage;import javax.naming.Context;import javax.naming.InitialContext;import javax.naming.NamingException;/* * 异步接收P2P消息 * * author liuyan * */public

23、class JbossASyncMessageReciveriver implements MessageListener /* * 异步接收 */public void reciveMessage() try String Connection_Factory = ConnectionFactory;Context context = getInitialContext();ConnectionFactory connectionFactory = (ConnectionFactory) context.lookup(Connection_Factory);Destination dest

24、= (Destination) context.lookup(jbossJMS);Connection connection = connectionFactory.createConnection();connection.start();Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);MessageConsumer reciver = session.createConsumer(dest);/ 为JMS消费者绑定监听器reciver.setMessageListener(this);/

25、等待50秒Thread.sleep(50000);session.close();connection.close(); catch (Exception e) e.printStackTrace();private Context getInitialContext() /重复/* * 实现此方法，每当监听到此目的有消息了，就执行此方法 */Overridepublic void onMessage(Message message) / TODO Auto-generated method stubTextMessage textMessage = (TextMessage) message

26、;String messageText;try messageText = textMessage.getText();System.out.println(textMessage： + textMessage);System.out.println(消息内容 + messageText); catch (JMSException e) / TODO Auto-generated catch blocke.printStackTrace();/* * param args */public static void main(String args) JbossASyncMessageReciv

27、eriver jbossSyncMessageReciveriver = new JbossASyncMessageReciveriver();jbossSyncMessageReciveriver.reciveMessage();异步消费需要实现MessageListener接口和它的onMessage()方法，接收到消息后在onMessage()方法中执行自己的相关逻辑。按照上例的顺序执行代码后，消费者控制台如下：textMessage：delegator-JBossMessage:PERSISTENT, deliveryId=0消息内容hello 速速textMessage：delega

28、tor-JBossMessage:PERSISTENT, deliveryId=1消息内容welcome to JMS消费者开启50秒后自动退出，这就是异步消费模式的效果。就是说，老娘等你（消息生产者）50秒，有消息，老娘就收着，没有！50秒后老娘就撤了，过了这村就没这店了啊！2.发布/订阅Pub-Sub消息Pub-Sub有点1对多的意思，一个消息生产者，多个消息消费者共同接收。首先我们在JBoss中配置一个消息主题，生产者实际上是对此消息主题发送消息的，消息消费者实际上也是对此主题的消息进行接收的。在此我们配置了一个名字叫做jbossJMSTopic的消息主题。因为JMS1.1的标准在代码上

29、已经完全屏蔽了PTP与Pub-Sub的差异性，所以我们上面的代码可以修改一下重复利用消息消费者修改目的Destination dest = (Destination) context.lookup(jbossJMSTopic);消息生产者修改目的Destination dest = (Destination) context.lookup(jbossJMSTopic);其他代码不同变更下面我们为了查看消息群发的效果，执行2遍消息消费者目的的代码，之后执行1遍消息生产者，看到的效果如下：2个消息消费者控制台其中一个消费者textMessage：delegator-JBossMessage:PER

30、SISTENT, deliveryId=0消息内容hello 速速另一个消费者textMessage：delegator-JBossMessage:PERSISTENT, deliveryId=0消息内容hello 速速这证明了2个消费者都接收到了此消息。当然，异步消费也和上面代码一样，就是替换一个目的地址就可以，底层的实现为开发者屏蔽了这些不同JMS信息模式的差异。5. JMS消息常用的属性配置JMS的消息再发送之前还可以设置一些消息属性1.传递方式：持久化保存消息sender.setDeliveryMode(DeliveryMode.PERSISTENT);不用持久化保存消息sender.

31、setDeliveryMode(DeliveryMode.NON_PERSISTENT);2.消息的有效期时间sender.setTimeToLive(2000);代表此消息2秒后就作废了，即使是将消息持久化到一个物理空间上，2秒钟后也会删除。如果不设置，默认是永远不会过期的。3.消息的优先级sender.setPriority(9);将此消息设置成优先级最大的9，相当于500里加急文书！。默认的优先级是4。当然之前所有做的一切属性设置，最后如果调用了MessageProducer.send(消息实体,存储模式 , 优先级, 有效时间);这种最终、比较全面的方法所覆盖。6. JMS消息过滤JM

32、S消息消费者可以有选择性的接收来自目的地的消息，在创建消费者接口MessageCunsumer的时候就可以指定接收消息的条件。比如接收优先级9的消息package jms.pts;import java.util.Properties;import javax.jms.Connection;import javax.jms.ConnectionFactory;import javax.jms.Destination;import javax.jms.MessageConsumer;import javax.jms.Session;import javax.jms.TextMessage;imp

33、ort javax.naming.Context;import javax.naming.InitialContext;import javax.naming.NamingException;/* * 同步接收优先级小于9的P2S消息 * * author liuyan * */public class JbossFilterSyncMessageReciveriver1 public void reciveMessage() try String Connection_Factory = ConnectionFactory;Context context = getInitialContex

34、t();ConnectionFactory connectionFactory = (ConnectionFactory) context.lookup(Connection_Factory);Destination dest = (Destination) context.lookup(jbossJMSTopic);Connection connection = connectionFactory.createConnection();connection.start();Session session = connection.createSession(false,Session.CLI

35、ENT_ACKNOWLEDGE);MessageConsumer reciver = session.createConsumer(dest,JMSPriority 9);TextMessage textMessage = (TextMessage) reciver.receive();String messageText = textMessage.getText();System.out.println(textMessage： + textMessage);System.out.println(消息内容 + messageText);/ textMessage.acknowledge()

36、;session.close();connection.close(); catch (Exception e) e.printStackTrace();private Context getInitialContext() /* * param args */public static void main(String args) JbossFilterSyncMessageReciveriver1 jbossSyncMessageReciveriver = new JbossFilterSyncMessageReciveriver1();jbossSyncMessageReciverive

37、r.reciveMessage();消息生产者如下package jms.pts;import java.util.Properties;import javax.jms.Connection;import javax.jms.ConnectionFactory;import javax.jms.DeliveryMode;import javax.jms.Destination;import javax.jms.MessageProducer;import javax.jms.Session;import javax.jms.TextMessage;import javax.naming.Co

38、ntext;import javax.naming.InitialContext;import javax.naming.NamingException;/* * P2S消息发送者 * * author liuyan * */public class JbossMessageSender /* * 发送消息 * * throws Exception */public void sendMessage() try String Connection_Factory = ConnectionFactory;Context context = getInitialContext();Connecti

39、onFactory connectionFactory = (ConnectionFactory) context.lookup(Connection_Factory);Destination dest = (Destination) context.lookup(jbossJMSTopic);Connection connection = connectionFactory.createConnection();Session session = connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);MessageProduce

40、r sender = session.createProducer(dest);sender.setDeliveryMode(DeliveryMode.PERSISTENT);/sender.setTimeToLive(2000);sender.setPriority(9);TextMessage textMessage = session.createTextMessage();textMessage.setStringProperty(ConType, txt);textMessage.setText(hello 速速);sender.send(textMessage);System.ou

41、t.println(停顿5秒钟);/ 等待50秒Thread.sleep(5000);System.out.println(停顿5秒钟完毕);textMessage.setText(welcome to JMS);sender.send(textMessage);session.close();connection.close(); catch (Exception e) e.printStackTrace();private Context getInitialContext() /* * param args * throws Exception */public static void

42、main(String args) throws Exception JbossMessageSender jbossMessageSender = new JbossMessageSender();jbossMessageSender.sendMessage();此消息发出后，消息消费者并不会从目的接收到该消息，因为不满足消费者的设定条件！消费者只对优先级小于9的消息敏感。同样的类似于消息类型为txtMessageConsumer reciver = session.createConsumer(dest, ConType=txt);7. JMS的事务控制JMS控制消息发送的事务有2种途径，

43、一个是利用Connection接口建立事务性会话，另一个就是利用JTA标准接口控制事务了。Session事务性会话public void sendMessage() try String Connection_Factory = ConnectionFactory;Context context = getInitialContext();ConnectionFactory connectionFactory = (ConnectionFactory) context.lookup(Connection_Factory);Destination dest = (Destination) con

44、text.lookup(jbossJMSTopic);Connection connection = connectionFactory.createConnection();Session session = connection.createSession(true,Session.CLIENT_ACKNOWLEDGE);MessageProducer sender = session.createProducer(dest);sender.setDeliveryMode(DeliveryMode.PERSISTENT);/ sender.setTimeToLive(2000);sende

45、r.setPriority(9);TextMessage textMessage = session.createTextMessage();textMessage.setStringProperty(ConType, txt);textMessage.setText(hello 速速);sender.send(textMessage);System.out.println(停顿5秒钟);/ 等待50秒Thread.sleep(5000);System.out.println(停顿5秒钟完毕);textMessage.setText(welcome to JMS);sender.send(te

46、xtMessage);/此处抛出运行时异常int a = 1/0;mit();session.close();connection.close(); catch (Exception e) e.printStackTrace();此代码会抛出异常java.lang.ArithmeticException: / by zero所以消费者是无法接收到此消息的，因为事务还没有提交呢，就抛出了异常，还有一点就是在建立session的时候如果是建立的事务性会话，必须手工代码让会话提交事务mit();消息才能送达到目的地！JTA管理JMS消息事务如果需要让JMS

47、、数据库、EJB操作等参与到事务当中，则应该考虑JTA事务。因为Session事务性会话总会打开一个新的事务，在Session中未commit之前，全部都在一个Session事务中，而数据库操作、EJB操作等等不能参与、共享此会话性事务中。所以此时应该用JTA来包装此事务。来看JTA控制的消息生产者的一个Servlet代码public void doPost(HttpServletRequest request, HttpServletResponse response)throws ServletException, IOException UserTransaction userTrans

48、action = null;java.sql.Connection connectionSQL = null;DataSource dataSource = null;Statement statement = null;Context context;try context = new InitialContext();dataSource = (DataSource) context.lookup(java:/jbossdemo);String Connection_Factory = ConnectionFactory;ConnectionFactory connectionFactor

49、y = (ConnectionFactory) context.lookup(Connection_Factory);Destination dest = (Destination) context.lookup(jbossJMSTopic);/ 获得JTA相关用户接口userTransaction = (UserTransaction) context.lookup(UserTransaction);/ 开始事务userTransaction.begin();javax.jms.Connection connection = connectionFactory.createConnectio

50、n();Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);MessageProducer sender = session.createProducer(dest);sender.setDeliveryMode(DeliveryMode.PERSISTENT);/ sender.setTimeToLive(2000);sender.setPriority(9);TextMessage textMessage = session.createTextMessage();textMessage.se

51、tStringProperty(ConType, txt);textMessage.setText(hello 第一个消息);sender.send(textMessage);System.out.println(停顿5秒钟);/ 等待50秒Thread.sleep(5000);System.out.println(停顿5秒钟完毕);textMessage.setText(welcome to JMS);sender.send(textMessage);/ 执行插入数据业务connectionSQL = dataSource.getConnection();statement = connectionSQL.createStatement();String insert1 = insert into person values(13,欧阳上智);String insert2 = insert into person v