Jms基础知识整理 JMSSpringBeanJBossActiveMQ

开始文章之前先澄清几个概念

什么是消息
消息是一个用于在组件和应用程序之间通讯的的方法。消息之间的传递是点对点的。任何终端之间都可以相互接受和发送消息。并且每个终端都必须遵守如下的规则
 -> 创建消息 -> 发送消息 -> 接收消息 -> 读取消息

为什么要使用消息
理由很简单，消息是一个分布式的低耦合通讯方案。A发送一个消息到一个agent ,B作为接受者去agent上获取消息。但是A,B不需要同时到agent上去注册。agent作为一个中转为A,B提供搞效率的通讯服务。

开发者的关注点
走到这里，我也不想去解释jms spec上那些抽象且复杂的概念了，说的很白，1年多了我自己也没弄懂是个什么东西，也没时间从头到尾去仔细的看，同时我认为没必要，我所关注的是如何让jms跑起来，并且工作正常，所以spec只是个字典，当我需要用的时候才去查。

开发者的jms环境
遵守简单明了的原则，所谓jms环境只是2个对象
1> ConnectionFactory
2> Destination
通常Provider会提供JNDI的对象获取，具体方法可以去Privider的网站上搜索jndi support

下面我以jbossMq为介质跑一个简单的jms，为了保证jms的本质清晰，我没有使用jbossMq的Api，而是直接调用的jms Api.

java 代码

 

1. package com.javaeye.jms.jboss;  
2.   
3. import javax.jms.Connection;  
4. import javax.jms.ConnectionFactory;  
5. import javax.jms.Destination;  
6. import javax.jms.JMSException;  
7. import javax.jms.MessageConsumer;  
8. import javax.jms.MessageProducer;  
9. import javax.jms.Queue;  
10. import javax.jms.QueueSender;  
11. import javax.jms.Session;  
12. import javax.jms.TextMessage;  
13. import javax.naming.Context;  
14. import javax.naming.InitialContext;  
15. import javax.naming.NamingException;  
16.   
17. public class JbossNativeJmsImpl {  
18.      
19.     /** 
20.      * @author zuly 
21.      * 
22.      * following jms ptp domain, use an simple text message to test 
23.      * 
24.      * A jms ptp sender will following the steps below! 
25.      *     1> get an ConnectionFactory use JNDI Lookup Or Initial it yourself 
26.      *     2> use this ConnectionFactory to start a jms connection
    
27.      *        [spec to jms 1.1 apito get the main idea of it ] 
28.      *     3> use connection to create a jms session 
29.      *     4> get a queue destination / messege agent 
30.      *     5> start the Producer[jms1.1 spec] by a session 
31.      *     6> get messege Object or initial it yourself by implements the messegeor 
    
32.      *        it's sub interfaces 
33.      *     7> call sender or send it selfing 
34.      *     8> finallized the connection object or it will throw a warning to you! 
35.      * 
36.      * @param messege 
37.      * @throws NamingException 
38.      * @throws JMSException 
39.      */  
40.     public void sendingProcessing(String messege) throws NamingException, JMSException{  
41.         Context ctx = new InitialContext();  
42.         ConnectionFactory cf = (ConnectionFactory) ctx.lookup("java:JmsXA");  
43.         Connection conn = cf.createConnection();  
44.         Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);  
45.         Destination dest = (Queue) ctx.lookup("queue/A");  
46.         MessageProducer msgp = session.createProducer(dest);  
47.         QueueSender sender = (QueueSender) msgp;  
48.         TextMessage msg = session.createTextMessage();  
49.         msg.setText(messege);  
50.         sender.send(msg);  
51.         conn.close();  
52.     }  
53.      
54.      
55.      
56.     /** 
57.      * @author zuly 
58.      * 
59.      * following jms ptp domain, use an simple text message to test 
60.      * 
61.      * A jms ptp retriver will following the steps below! 
62.      *     1> get an ConnectionFactory use JNDI Lookup Or Initial it yourself 
63.      *     2> use this ConnectionFactory to start a jms connection 
64.      *        [spec to jms 1.1 api to get the main idea of it ] 
65.      *     3> use connection to create a jms session 
66.      *     4> get a queue destination / messege agent 
67.      *     5> retrive a consumer from session 
68.      *     6> start the jms connection to retrivte the message 
69.      *     7> get message from consumer 
70.      *  
71.      * @return textMessege 
72.      * @throws NamingException 
73.      * @throws JMSException 
74.      */  
75.     public String retriveingProcessing() throws NamingException, JMSException{  
76.         Context ctx = new InitialContext();  
77.         ConnectionFactory cf = (ConnectionFactory) ctx.lookup("java:JmsXA");  
78.         Connection conn = cf.createConnection();  
79.         Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);  
80.         Destination dest = (Queue) ctx.lookup("queue/A");  
81.         MessageConsumer msgconsumer = session.createConsumer(dest);  
82.         //MessageListener ml = new JmsListenner();  
83.         //msgconsumer.setMessageListener(ml);  
84.         conn.start();  
85.         TextMessage msg = (TextMessage) msgconsumer.receive();  
86.         conn.close();  
87.         System.out.println("messege is" + msg.getText());  
88.         return msg.getText();  
89.     }  
90. }  

<point to="" point=""></point> <point to="" point=""></point>

注意retrive函数中comment的掉的两行，消息Listener的作用是实现异步通讯，但是它有一个约定，必须和发送者
保持物理上的分离，针对于jboss而言，就要求这个Listener必须跑在容器外面。这是一个很搞的问题，每天Jms的邮件列表里面都有无数的这样的问题发过来。但是回复的人很少。我自己也从来不回复。 其实我也不清楚写这篇文章到底是出于什么目的，怕只是让这么一个简单的问题有一个回答而已。

把下面这个程序跑起来就可以异步接受消息了。

java 代码

 

1. package com.javaeye.jms.jboss;  
2.   
3. import java.util.Properties;  
4.   
5. import javax.jms.Connection;  
6. import javax.jms.ConnectionFactory;  
7. import javax.jms.Destination;  
8. import javax.jms.JMSException;  
9. import javax.jms.MessageConsumer;  
10. import javax.jms.MessageListener;  
11. import javax.jms.Session;  
12. import javax.naming.Context;  
13. import javax.naming.InitialContext;  
14. import javax.naming.NamingException;  
15.   
16. import com.javaeye.spring.services.jms.mdp.JmsListenner;  
17.   
18. public class JbossJmsAsync {  
19.   
20.     /** 
21.      * @param args 
22.      * @throws NamingException  
23.      * @throws JMSException  
24.      */  
25.     public static void main(String[] args) throws NamingException, JMSException {  
26.         Properties pops = new Properties();  
27.         pops.setProperty("jboss.bind.address", "0.0.0.0");  
28.         pops.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");  
29.         pops.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");  
30.         pops.setProperty("java.naming.provider.url", "localhost");  
31.         Context ctx = new InitialContext(pops);  
32.         ConnectionFactory cf = (ConnectionFactory) ctx.lookup("ConnectionFactory");  
33.         Connection conn = cf.createConnection();  
34.         Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);  
35.         Destination dest = (Destination) ctx.lookup("queue/A");  
36.         MessageConsumer msgConsumer = session.createConsumer(dest);  
37.         MessageListener ml = new JmsListenner();  
38.         msgConsumer.setMessageListener(ml);   
39.         conn.start();  
40.     }  
41.   
42. }  

javaeye的主题好像是spring，为了迎合领导，下面我把这套东西跑在spring里面。同时我发现spring对jms的包装真的简单，而且还提供了一个模版，虽然这个模版的接口是在是很罗唆。

ps:今天是第1次用spring在reference里找了半天找不到方法注入的办法，于是google了一个注入办法，不合理的地方请大家指出。首先我通过方法来注入ConnectionFactory和Destination这两个对象来支撑jms环境

java 代码

 

1. package com.javaeye.spring.services.jms.mdp;  
2.   
3. import java.util.Properties;  
4.   
5. import javax.jms.ConnectionFactory;  
6. import javax.jms.Destination;  
7. import javax.jms.Queue;  
8. import javax.naming.Context;  
9. import javax.naming.InitialContext;  
10. import javax.naming.NamingException;  
11.   
12. public class UserJmsTransactionUtil {  
13.   
14.     private String connectionFactoryJndiLookUp;  
15.      
16.     private String destinationJndiLookUp;  
17.      
18.     private String localConnectionFactoryJndiLookUp;  
19.      
20.     private String containerType;  
21.      
22.      
23.     public String getConnectionFactoryJndiLookUp() {  
24.         return connectionFactoryJndiLookUp;  
25.     }  
26.   
27.   
28.   
29.     public void setConnectionFactoryJndiLookUp(String connectionFactoryJndiLookUp) {  
30.         this.connectionFactoryJndiLookUp = connectionFactoryJndiLookUp;  
31.     }  
32.   
33.   
34.   
35.     public String getDestinationJndiLookUp() {  
36.         return destinationJndiLookUp;  
37.     }  
38.   
39.   
40.   
41.     public void setDestinationJndiLookUp(String destinationJndiLookUp) {  
42.         this.destinationJndiLookUp = destinationJndiLookUp;  
43.     }  
44.   
45.   
46.   
47.     public ConnectionFactory getConnectionFactory() throws NamingException{  
48.         Context ctx = new InitialContext();  
49.         ConnectionFactory cf = (ConnectionFactory) ctx.lookup(connectionFactoryJndiLookUp);  
50.         return cf;  
51.     }  
52.      
53.      
54.     public Destination getJmsDestination() throws NamingException{  
55.         Context ctx = new InitialContext();  
56.         Destination dest = (Queue) ctx.lookup(destinationJndiLookUp);  
57.         return dest;  
58.     }  
59.      
60.      
61.     public ConnectionFactory getQueueConnectionFactory() throws NamingException{  
62.         Properties pops = new Properties();  
63.         pops.setProperty("jboss.bind.address", "0.0.0.0");  
64.         pops.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");  
65.         pops.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");  
66.         pops.setProperty("java.naming.provider.url", "localhost");  
67.         Context ctx = new InitialContext(pops);  
68.         ConnectionFactory cf = (ConnectionFactory) ctx.lookup(localConnectionFactoryJndiLookUp);  
69.         return cf;  
70.     }  
71.      
72.      
73.     public Destination getLocalJmsDestination() throws NamingException{  
74.         Properties pops = new Properties();  
75.         pops.setProperty("jboss.bind.address", "0.0.0.0");  
76.         pops.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");  
77.         pops.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");  
78.         pops.setProperty("java.naming.provider.url", "localhost");  
79.         Context ctx = new InitialContext(pops);  
80.         Destination dest = (Destination) ctx.lookup(destinationJndiLookUp);  
81.         return dest;  
82.     }  
83.   
84.   
85.   
86.     public String getLocalConnectionFactoryJndiLookUp() {  
87.         return localConnectionFactoryJndiLookUp;  
88.     }  
89.   
90.   
91.   
92.     public void setLocalConnectionFactoryJndiLookUp(  
93.             String localConnectionFactoryJndiLookUp) {  
94.         this.localConnectionFactoryJndiLookUp = localConnectionFactoryJndiLookUp;  
95.     }     
96. }
    

发送端的配置如下

xml 代码

 

1. <beans>  
2.     <bean id="userJmsUtil" class="com.javaeye.spring.services.jms.mdp.UserJmsTransactionUtil">  
3.         <property name="connectionFactoryJndiLookUp" value="java:JmsXA"><!---->property>  
4.         <property name="destinationJndiLookUp" value="queue/A"><!---->property>  
5.         <property name="localConnectionFactoryJndiLookUp" value="ConnectionFactory"><!---->property>  
6.     <!---->bean>  
7.   
8.     <bean id="connectionFactory" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">  
9.         <property name="targetObject" ref="userJmsUtil"><!---->property>  
10.         <property name="targetMethod" value="getConnectionFactory"><!---->property>  
11.     <!---->bean>  
12.       
13.     <bean id="queue" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">  
14.         <property name="targetObject" ref="userJmsUtil"><!---->property>  
15.         <property name="targetMethod" value="getJmsDestination"><!---->property>  
16.     <!---->bean>  
17.           
18.     <bean id="jmsQueue" class="org.springframework.jms.core.JmsTemplate">  
19.         <property name="connectionFactory" ref="connectionFactory"><!---->property>  
20.         <property name="defaultDestination" ref="queue"><!---->property>  
21.         <property name="messageConverter">  
22.             <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"><!---->bean>  
23.         <!---->property>  
24.     <!---->bean>  
25. <!---->beans>  

ps:javaeye的模版工具bug还真多,不管了.

如果使用Listenner的化,一样需要遵守发送者和接收者物理隔离的原则,我的做法是把发送者配到一个xml中,在把接受者配到另外一个xml中去,发送的配置绑定到容器里,接收者的跑在本地.否则spring初始化是过不去的.

下面这个程序是发送消息的程序.使用了spring的模版,发条消息比new个对象还简单.同时spring还提供了适配器的接口,一样通过声明式的配置,这样可以在同一个接口里发送各种类型的消息了.同时支持事务,我还不知道这个有什么用呵呵,第1次使用嘛!但是就使用上来说,spring是最简单的.2者都只需要注入一个对象而已.

java 代码

 

1. @Test public void send(){  
2.     ApplicationContext ac = new FileSystemXmlApplicationContext("jms.xml");  
3.     BeanFactory bf = ac;  
4.     JmsTemplate jt = (JmsTemplate) bf.getBean("jmsQueue");  
5.     jt.convertAndSend("2132134");  
6. }  

接收端的配置如下

xml 代码

 

1. xml version="1.0" encoding="UTF-8"?>  
2. >  
3. <beans>  
4.   
5.     <bean id="listenner" class="com.javaeye.spring.services.jms.mdp.JmsListenner"><!---->bean>  
6.       
7.     <bean id="userJmsUtil" class="com.javaeye.spring.services.jms.mdp.UserJmsTransactionUtil">  
8.         <property name="connectionFactoryJndiLookUp" value="java:JmsXA"><!---->property>  
9.         <property name="destinationJndiLookUp" value="queue/A"><!---->property>  
10.         <property name="localConnectionFactoryJndiLookUp" value="ConnectionFactory"><!---->property>  
11.     <!---->bean>  
12.   
13.     <bean id="localConnectionFactory" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">  
14.         <property name="targetObject" ref="userJmsUtil"><!---->property>  
15.         <property name="targetMethod" value="getQueueConnectionFactory"><!---->property>  
16.     <!---->bean>  
17.       
18.     <bean id="localDestination" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">  
19.         <property name="targetObject" ref="userJmsUtil"><!---->property>  
20.         <property name="targetMethod" value="getLocalJmsDestination"><!---->property>  
21.     <!---->bean>  
22.       
23.     <bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
24.         <property name="concurrentConsumers" value="5"><!---->property>  
25.         <property name="connectionFactory" ref="localConnectionFactory"><!---->property>  
26.         <property name="destination" ref="localDestination"><!---->property>  
27.         <property name="messageListener" ref="listenner"><!---->property>  
28.     <!---->bean>  
29. <!---->beans>  

接收端由于需要从jbossmq里取ConnectionFactory和Destination,所以,我调用的是userJmsUtil的localLookup.这个函数的作用等同于发送者的那个函数,只不过前者是容器外获取,而后者是容器内的而已.

java 代码

 

1. package com.javaeye.spring.services.jms.mdp;  
2.   
3. import javax.jms.JMSException;  
4. import javax.jms.Message;  
5. import javax.jms.MessageListener;  
6. import javax.jms.TextMessage;  
7.   
8. public class JmsListenner implements MessageListener {  
9.   
10.     public void onMessage(Message message) {  
11.         try {  
12.             TextMessage msg = (TextMessage) message;  
13.             System.out.println(msg.getText());  
14.         } catch (JMSException e) { e.printStackTrace(); }  
15.     }  
16.   
17. } 
    

spring对jms的整合里提到了一个jms provider ActiveMQ,要用一个开源框架要做的第一件事就是先跑一个demo起来,同样,我们要做的事还是获取ConnectionFactory和Destination对象,还好,ActiveMQ的JNDI实现比jbossMQ还要简单,直接通过一个本地的Context就可以查到了,具体的可以参照ActiveMQ官方的支持文档.