开始文章之前先澄清几个概念
什么是消息
消息是一个用于在组件和应用程序之间通讯的的方法。消息之间的传递是点对点的。任何终端之间都可以相互接受和发送消息。并且每个终端都必须遵守如下的规则
-> 创建消息 -> 发送消息 -> 接收消息 -> 读取消息
为什么要使用消息
理由很简单,消息是一个分布式的低耦合通讯方案。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 代码
- package com.javaeye.jms.jboss;
-
- import javax.jms.Connection;
- import javax.jms.ConnectionFactory;
- import javax.jms.Destination;
- import javax.jms.JMSException;
- import javax.jms.MessageConsumer;
- import javax.jms.MessageProducer;
- import javax.jms.Queue;
- import javax.jms.QueueSender;
- import javax.jms.Session;
- import javax.jms.TextMessage;
- import javax.naming.Context;
- import javax.naming.InitialContext;
- import javax.naming.NamingException;
-
- public class JbossNativeJmsImpl {
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- public void sendingProcessing(String messege) throws NamingException, JMSException{
- Context ctx = new InitialContext();
- ConnectionFactory cf = (ConnectionFactory) ctx.lookup("java:JmsXA");
- Connection conn = cf.createConnection();
- Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Destination dest = (Queue) ctx.lookup("queue/A");
- MessageProducer msgp = session.createProducer(dest);
- QueueSender sender = (QueueSender) msgp;
- TextMessage msg = session.createTextMessage();
- msg.setText(messege);
- sender.send(msg);
- conn.close();
- }
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- public String retriveingProcessing() throws NamingException, JMSException{
- Context ctx = new InitialContext();
- ConnectionFactory cf = (ConnectionFactory) ctx.lookup("java:JmsXA");
- Connection conn = cf.createConnection();
- Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Destination dest = (Queue) ctx.lookup("queue/A");
- MessageConsumer msgconsumer = session.createConsumer(dest);
-
-
- conn.start();
- TextMessage msg = (TextMessage) msgconsumer.receive();
- conn.close();
- System.out.println("messege is" + msg.getText());
- return msg.getText();
- }
- }
<point to="" point=""></point>
<point to="" point=""></point>
注意retrive函数中comment的掉的两行,消息Listener的作用是实现异步通讯,但是它有一个约定,必须和发送者
保持物理上的分离,针对于jboss而言,就要求这个Listener必须跑在容器外面。这是一个很搞的问题,每天Jms的邮件列表里面都有无数的这样的问题发过来。但是回复的人很少。我自己也从来不回复。 其实我也不清楚写这篇文章到底是出于什么目的,怕只是让这么一个简单的问题有一个回答而已。
把下面这个程序跑起来就可以异步接受消息了。
java 代码
- package com.javaeye.jms.jboss;
-
- import java.util.Properties;
-
- import javax.jms.Connection;
- import javax.jms.ConnectionFactory;
- import javax.jms.Destination;
- import javax.jms.JMSException;
- import javax.jms.MessageConsumer;
- import javax.jms.MessageListener;
- import javax.jms.Session;
- import javax.naming.Context;
- import javax.naming.InitialContext;
- import javax.naming.NamingException;
-
- import com.javaeye.spring.services.jms.mdp.JmsListenner;
-
- public class JbossJmsAsync {
-
-
-
-
-
-
- public static void main(String[] args) throws NamingException, JMSException {
- Properties pops = new Properties();
- pops.setProperty("jboss.bind.address", "0.0.0.0");
- pops.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");
- pops.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");
- pops.setProperty("java.naming.provider.url", "localhost");
- Context ctx = new InitialContext(pops);
- ConnectionFactory cf = (ConnectionFactory) ctx.lookup("ConnectionFactory");
- Connection conn = cf.createConnection();
- Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Destination dest = (Destination) ctx.lookup("queue/A");
- MessageConsumer msgConsumer = session.createConsumer(dest);
- MessageListener ml = new JmsListenner();
- msgConsumer.setMessageListener(ml);
- conn.start();
- }
-
- }
javaeye的主题好像是spring,为了迎合领导,下面我把这套东西跑在spring里面。同时我发现spring对jms的包装真的简单,而且还提供了一个模版,虽然这个模版的接口是在是很罗唆。
ps:今天是第1次用spring在reference里找了半天找不到方法注入的办法,于是google了一个注入办法,不合理的地方请大家指出。首先我通过方法来注入ConnectionFactory和Destination这两个对象来支撑jms环境
java 代码
- package com.javaeye.spring.services.jms.mdp;
-
- import java.util.Properties;
-
- import javax.jms.ConnectionFactory;
- import javax.jms.Destination;
- import javax.jms.Queue;
- import javax.naming.Context;
- import javax.naming.InitialContext;
- import javax.naming.NamingException;
-
- public class UserJmsTransactionUtil {
-
- private String connectionFactoryJndiLookUp;
-
- private String destinationJndiLookUp;
-
- private String localConnectionFactoryJndiLookUp;
-
- private String containerType;
-
-
- public String getConnectionFactoryJndiLookUp() {
- return connectionFactoryJndiLookUp;
- }
-
-
-
- public void setConnectionFactoryJndiLookUp(String connectionFactoryJndiLookUp) {
- this.connectionFactoryJndiLookUp = connectionFactoryJndiLookUp;
- }
-
-
-
- public String getDestinationJndiLookUp() {
- return destinationJndiLookUp;
- }
-
-
-
- public void setDestinationJndiLookUp(String destinationJndiLookUp) {
- this.destinationJndiLookUp = destinationJndiLookUp;
- }
-
-
-
- public ConnectionFactory getConnectionFactory() throws NamingException{
- Context ctx = new InitialContext();
- ConnectionFactory cf = (ConnectionFactory) ctx.lookup(connectionFactoryJndiLookUp);
- return cf;
- }
-
-
- public Destination getJmsDestination() throws NamingException{
- Context ctx = new InitialContext();
- Destination dest = (Queue) ctx.lookup(destinationJndiLookUp);
- return dest;
- }
-
-
- public ConnectionFactory getQueueConnectionFactory() throws NamingException{
- Properties pops = new Properties();
- pops.setProperty("jboss.bind.address", "0.0.0.0");
- pops.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");
- pops.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");
- pops.setProperty("java.naming.provider.url", "localhost");
- Context ctx = new InitialContext(pops);
- ConnectionFactory cf = (ConnectionFactory) ctx.lookup(localConnectionFactoryJndiLookUp);
- return cf;
- }
-
-
- public Destination getLocalJmsDestination() throws NamingException{
- Properties pops = new Properties();
- pops.setProperty("jboss.bind.address", "0.0.0.0");
- pops.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");
- pops.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");
- pops.setProperty("java.naming.provider.url", "localhost");
- Context ctx = new InitialContext(pops);
- Destination dest = (Destination) ctx.lookup(destinationJndiLookUp);
- return dest;
- }
-
-
-
- public String getLocalConnectionFactoryJndiLookUp() {
- return localConnectionFactoryJndiLookUp;
- }
-
-
-
- public void setLocalConnectionFactoryJndiLookUp(
- String localConnectionFactoryJndiLookUp) {
- this.localConnectionFactoryJndiLookUp = localConnectionFactoryJndiLookUp;
- }
- }
发送端的配置如下
xml 代码
- <beans>
- <bean id="userJmsUtil" class="com.javaeye.spring.services.jms.mdp.UserJmsTransactionUtil">
- <property name="connectionFactoryJndiLookUp" value="java:JmsXA"><!---->property>
- <property name="destinationJndiLookUp" value="queue/A"><!---->property>
- <property name="localConnectionFactoryJndiLookUp" value="ConnectionFactory"><!---->property>
- <!---->bean>
-
- <bean id="connectionFactory" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
- <property name="targetObject" ref="userJmsUtil"><!---->property>
- <property name="targetMethod" value="getConnectionFactory"><!---->property>
- <!---->bean>
-
- <bean id="queue" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
- <property name="targetObject" ref="userJmsUtil"><!---->property>
- <property name="targetMethod" value="getJmsDestination"><!---->property>
- <!---->bean>
-
- <bean id="jmsQueue" class="org.springframework.jms.core.JmsTemplate">
- <property name="connectionFactory" ref="connectionFactory"><!---->property>
- <property name="defaultDestination" ref="queue"><!---->property>
- <property name="messageConverter">
- <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"><!---->bean>
- <!---->property>
- <!---->bean>
- <!---->beans>
ps:javaeye的模版工具bug还真多,不管了.
如果使用Listenner的化,一样需要遵守发送者和接收者物理隔离的原则,我的做法是把发送者配到一个xml中,在把接受者配到另外一个xml中去,发送的配置绑定到容器里,接收者的跑在本地.否则spring初始化是过不去的.
下面这个程序是发送消息的程序.使用了spring的模版,发条消息比new个对象还简单.同时spring还提供了适配器的接口,一样通过声明式的配置,这样可以在同一个接口里发送各种类型的消息了.同时支持事务,我还不知道这个有什么用呵呵,第1次使用嘛!但是就使用上来说,spring是最简单的.2者都只需要注入一个对象而已.
java 代码
- @Test public void send(){
- ApplicationContext ac = new FileSystemXmlApplicationContext("jms.xml");
- BeanFactory bf = ac;
- JmsTemplate jt = (JmsTemplate) bf.getBean("jmsQueue");
- jt.convertAndSend("2132134");
- }
接收端的配置如下
xml 代码
- xml version="1.0" encoding="UTF-8"?>
- >
- <beans>
-
- <bean id="listenner" class="com.javaeye.spring.services.jms.mdp.JmsListenner"><!---->bean>
-
- <bean id="userJmsUtil" class="com.javaeye.spring.services.jms.mdp.UserJmsTransactionUtil">
- <property name="connectionFactoryJndiLookUp" value="java:JmsXA"><!---->property>
- <property name="destinationJndiLookUp" value="queue/A"><!---->property>
- <property name="localConnectionFactoryJndiLookUp" value="ConnectionFactory"><!---->property>
- <!---->bean>
-
- <bean id="localConnectionFactory" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
- <property name="targetObject" ref="userJmsUtil"><!---->property>
- <property name="targetMethod" value="getQueueConnectionFactory"><!---->property>
- <!---->bean>
-
- <bean id="localDestination" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
- <property name="targetObject" ref="userJmsUtil"><!---->property>
- <property name="targetMethod" value="getLocalJmsDestination"><!---->property>
- <!---->bean>
-
- <bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
- <property name="concurrentConsumers" value="5"><!---->property>
- <property name="connectionFactory" ref="localConnectionFactory"><!---->property>
- <property name="destination" ref="localDestination"><!---->property>
- <property name="messageListener" ref="listenner"><!---->property>
- <!---->bean>
- <!---->beans>
接收端由于需要从jbossmq里取ConnectionFactory和Destination,所以,我调用的是
userJmsUtil的localLookup.这个函数的作用等同于发送者的那个函数,只不过前者是容器外获取,而后者是容器内的而已.
java 代码
- package com.javaeye.spring.services.jms.mdp;
-
- import javax.jms.JMSException;
- import javax.jms.Message;
- import javax.jms.MessageListener;
- import javax.jms.TextMessage;
-
- public class JmsListenner implements MessageListener {
-
- public void onMessage(Message message) {
- try {
- TextMessage msg = (TextMessage) message;
- System.out.println(msg.getText());
- } catch (JMSException e) { e.printStackTrace(); }
- }
-
- }
spring对jms的整合里提到了一个jms provider ActiveMQ,要用一个开源框架要做的第一件事就是先跑一个demo起来,同样,我们要做的事还是获取ConnectionFactory和Destination对象,还好,ActiveMQ的JNDI实现比jbossMQ还要简单,直接通过一个本地的Context就可以查到了,具体的可以参照ActiveMQ官方的支持文档.