JMS与MQ的几种结合方式

本文来自CSDN博客，转载请标明出处：http://blog.csdn.net/guozhanxian/archive/2007/01/21/1489309.aspx

 

配置JNDI
用JMS实现消息的发送和接收时，经常会用到JNDI。因为JNDI这种方式比较灵活，对于编程也比较简单。
在安装了MQSeries Client for
Java之后，在\java\bin目录下找到JMSAdmin.config文件。该文件主要用来说明Context的存储方式及存储地址，对应于文件中的两个参数INITIAL_CONTEXT_FACTORY和PROVIDER_URL。典型的JMSAdmin.config文件内容如下：
#INITIAL_CONTEXT_FACTORY=com.sun.jndi.ldap.LdapCtxFactory
INITIAL_CONTEXT_FACTORY=com.sun.jndi.fscontext.RefFSContextFactory
#INITIAL_CONTEXT_FACTORY=com.ibm.ejs.ns.jndi.CNInitialContextFactory
#
#PROVIDER_URL=ldap://polaris/o=ibm,c=us
PROVIDER_URL=file:/d:/temp
#PROVIDER_URL=iiop://localhost/
#
SECURITY_AUTHENTICATION=none
INITIAL_CONTEXT_FACTORY表示JMSAdmin
Tool使用的服务提供商。当前有三种受支持的值。com.sun.jndi.ldap.LdapCtxFactory用于LDAP，如果使用它就必须安装一个LDAP服务器。com.sun.jndi.fscontext.RefFSContextFactory用于文件系统上下文，它只需要使用者提供存放上下文的文件路径。com.ibm.ejs.ns.jndi.CNInitialContextFactory是专门为websphere提供的，它需要和websphere的CosNaming资源库一起使用。
PROVIDER_URL表示会话初始上下文的URL，由JMSAdmin
tool实现的所有JNDI操作的根。它和INITIAL_CONTEXT_FACTORY一一对应。
ldap://hostname/contextname 用于LDAP
file:[drive:]/pathname 用于文件系统上下文
iiop://hostname[:port]/[?TargetContext=ctx] 用于访问websphere
CosNaming名称空间
最后还有一个参数SECURITY_AUTHENTICATION，用于说明JNDI是否把安全性凭证传递给了您使用的服务供应商。只有当使用了LDAP服务供应商时，才使用此参数。此参数有三个值，none（匿名认证）、simple（简单认证）和CRAM-MD5认证机制。如果没有提供有效值，缺省值为none。
确认配置文件之后，可以在\java\bin目录下启动JMSAdmin控制台。也可以在任何目录下用下面的命令来启动控制台：
JMSAdmin -cfg MQ_JAVA_INSTALL_PATH\java\bin\JMSAdmin.config
其中MQ_JAVA_INSTALL_PATH为MQSeries Client for Java安装的根目录。
若启动失败，则好好检查一下您的环境变量是否设置正确。根据我个人的经验，除了把com.ibm.mq.jar和com.ibm.mqjms.jar加入到环境变量外，还要把fscontext.jar和providerutil.jar加入到环境变量。
进入JMSAdmin控制台后，您可以自由定义sub context。对于子上下文的操作，主要有一下命令：
display ctx
define ctx(ctxname)
change ctx(ctxname)
change ctx(=up)
change ctx(=init)
delete ctx(ctxname)
当然，在这里的主要任务并非是用来定义sub context，而是用来定义以下几个对象：
MQQueueConnectionFactory
MQTopicConnectionFactory
MQQueue
MQTopic
（还有其它的一些对象，如MQXAQueueConnectionFactory等，不常用到，在此不作说明。）
可以使用很多动词来操纵目录名称空间中的受管理对象。ALTER、DEFINE、DISPLAY、DELETE、COPY和MOVE，它们的用法都算比较简单，这里只列举一二以作说明。
例一：定义一QueueConnectionFactory，连接主机10.10.10.18，端口1414
DEFINE QCF(EXAMPLEQCF)+
DESC(Example Queue Connection Factory)+
TRAN(CLIENT)+
HOST(10.10.10.18)+
QMGR(QM_EXAMPLE)+
CHAN(S_EXAMPLE)+
PORT(1414)+
CCSID(1381)
例二：定义一Queue，其对应于MQ中的Q_EXAMPLE
DEFINE Q(EXAMPLEQL)+
DESC(Local queue)+
QMGR(QM_EXAMPLE)+
QUEUE(Q_EXAMPLE)+
CCSID(1381)

 

用JMS实现MQ编程
上面我们说明了怎样用JMSAdmin
Tool定义MQ对象的上下文。我们的最终目的是要用JMS来实现MQ编程，以实现在程序中对MQ队列进行收、发消息。所以，下面我们将重点讨论一下MQ的JMS实现。
如果您对JMS编程很熟悉，那么您也就会用JMS来实现MQ编程，因为用JMS来编写MQ程序与编写一般的JMS程序没有太大的差别。举个例子，当我们想发送一条消息到MQ的队列中，再从该队列中取回消息时，我们编程时主要有四个步骤。首先我们要初始化在程序中要用到的对象，然后才可以发送消息到队列中去，再就是收取消息了，最后要清除那些永久对象。这些都和普通的JMS程序相当。程序的源代码如下：

代码

import java.util.Hashtable;
import javax.jms.*;
import javax.naming.*;
import javax.naming.directory.*;
public class sample ...{
protected QueueConnectionFactory factory=null;
protected QueueConnection connection;
protected QueueSession queueSession;
protected TextMessage outMessage;
protected QueueSender queueSender;
protected QueueReceiver queueReceiver;
public static final String qcfLookup="EXAMPLEQCF";
public static final String qLookup="EXAMPLEQL";
public static final String icf =
"com.sun.jndi.fscontext.RefFSContextFactory";
public String url ="file:/d:/temp";
public void sampleInit() throws Exception ...{
Hashtable environment = new Hashtable();
environment.put(Context.INITIAL_CONTEXT_FACTORY, icf);
environment.put(Context.PROVIDER_URL, url);
environment.put(Context.REFERRAL, "throw");
Context ctx=new InitialDirContext(environment);
factory = (QueueConnectionFactory)ctx.lookup(qcfLookup);
Queue q1=null;
q1=(Queue)ctx.lookup(qLookup);
connection = factory.createQueueConnection();
queueSession = connection.createQueueSession(false,
Session.AUTO_ACKNOWLEDGE);
queueSender = queueSession.createSender(q1);
queueSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
outMessage = queueSession.createTextMessage();
queueReceiver = queueSession.createReceiver(q1);
connection.start();
}
public void sendMessageOut(String message) throws JMSException ...{
outMessage.setText(message);
queueSender.send(outMessage);
}
public String receiveMessage() throws Exception...{
return ((TextMessage)queueReceiver.receive()).getText();
}
public void sampleClose() throws JMSException ...{
queueSession.close();
connection.close();
}
public static void main(String[] args)...{
String rec;
sample sp = new sample();
try ...{
sp.sampleInit();
sp.sendMessageOut("Hello World!");
java.lang.Thread.sleep(4000);
rec=sp.receiveMessage();
System.out.println("Receive text is : "+rec);
sp.sampleClose();
}catch(Exception e) ...{
e.printStackTrace();
}
}
}

 

 

理解示例类

　　如果你编写过JMS应用程序，就很容易理解JNDIUtil 和 Tester示例类(从http://assets.devx.com/sourcecode/WebSphereMQ_JMSSource&Classes.zip下载Java源文件和编译好的类)，你创建的JMS管理对象隐藏了所有厂家专利实现。

　　JNDIUtil类

　　JNDIUtil包括使用名字通过JNDI查找检索对象的方法，参考清单1，你可以使用这个类中的方法检索你在MQ管理器中定义的JMS对象的引用情况。

　　清单1 JNDIUtil.java

package devx.articles.mqjms;

　　// JMS 类

　　import javax.jms.JMSException;

　　import javax.jms.Queue;

　　import javax.jms.QueueConnectionFactory;

　　// JNDI 类

　　import javax.naming.InitialContext;

　　import javax.naming.Context;

　　import javax.naming.NamingException;

　　// 标准 Java类

　　import java.util.Hashtable;

　　/**

　　*

　　* A wrapper class for JNDI calls

　　*

　　*/

　　public class JNDIUtil

　　{

　　private Context context;

　　public JNDIUtil(String icf, String url) throws JMSException, NamingException

　　{

　　Hashtable environment = new Hashtable();

　　environment.put(Context.INITIAL_CONTEXT_FACTORY, icf );

　　environment.put(Context.PROVIDER_URL, url);

　　context= new InitialContext( environment );

　　}

　　/**

　　*

　　* @param ObjName Object Name to be retrieved

　　* @return Retrieved Object

　　* @throws NamingException

　　*/

　　private Object getObjectByName(String ObjName) throws NamingException

　　{

　　return context.lookup( ObjName );

　　}

　　/**

　　* A convenience method that returns QueueConnectionFactory objects (no casting required)

　　* @param factoryName QueueConnectionFactory JNDI name

　　* @return QueueConnectionFactory object

　　* @throws NamingException

　　*/

　　public QueueConnectionFactory getQueueConnectionFactory(String factoryName) throws NamingException

　　{

　　return (QueueConnectionFactory) getObjectByName(factoryName);

　　}

　　/**

　　* A convenience method that returns Queue objects (no casting required)

　　* @param queueName

　　* @return

　　* @throws NamingException

　　*/

　　public Queue getQueue(String queueName) throws NamingException

　　{

　　return (Queue) getObjectByName(queueName);

　　}

　　}　　　　Tester类

　　Tester类向OUT.QUEUE中写入消息，从IN.QUEUE中读取消息。参考清单2.

　　清单2 Tester.java

　package devx.articles.mqjms;

　　//JMS 类

　　import javax.jms.Queue;

　　import javax.jms.QueueSession;

　　import javax.jms.QueueConnection;

　　import javax.jms.QueueConnectionFactory;

　　import javax.jms.JMSException;

　　import javax.jms.Session;

　　import javax.jms.QueueSender;

　　import javax.jms.QueueReceiver;

　　import javax.jms.TextMessage;

　　import javax.jms.Message;

　　//JNDI 类

　　import javax.naming.NamingException;

　　//标准 Java 类

　　/**

　　*

　　* A class to test JMS with IBM MQ

　　*

　　*/

　　public class Tester

　　{

　　public static String icf = "com.sun.jndi.fscontext.RefFSContextFactory";

　　public static String url = "file:/C:/JNDI-Directory";

　　public static void main(String[] vars) throws JMSException, NamingException

　　{

　　QueueSession session = null;

　　QueueConnection connection = null;

　　QueueConnectionFactory factory = null;

　　QueueSender queueSender = null;

　　QueueReceiver queueReceiver= null;

　　Queue oQueue = null; // 消息发送到的队列

　　Queue iQueue = null; // 接收消息的队列

　　try

　　{

　　JNDIUtil jndiUtil= new JNDIUtil(icf,url);

　　factory= jndiUtil.getQueueConnectionFactory("TestQM_QCF");

　　connection = factory.createQueueConnection();

　　// 启动(或重新启动)入站消息的连接地址，如果没有这个调用消息不会被接收

　　connection.start();

　　//表示一个非相互操作会话

　　boolean transacted = false;

　　session = connection.createQueueSession( transacted, Session.AUTO_ACKNOWLEDGE);

　　oQueue= jndiUtil.getQueue("OutputTestQueue");

　　queueSender = session.createSender(oQueue);

　　TextMessage oMsg = session.createTextMessage();

　　oMsg.setText("www.devx.com");

　　// 你还可以设置其他消息属性

　　queueSender.send(oMsg);

　　iQueue= jndiUtil.getQueue("InputTestQueue");

　　queueReceiver = session.createReceiver(iQueue);

　　Message iMsg = queueReceiver.receive(1000);

　　if ( iMsg != null )

　　System.out.println( ((TextMessage)iMsg).getText() );

　　else

　　System.out.println( "No messages in queue " );

　　}

　　finally

　　{

　　//总是释放资源

　　if ( queueReceiver!= null )

　　queueReceiver.close();

　　if ( queueSender!= null )

　　queueSender.close();

　　if ( session!= null )

　　session.close();

　　if ( connection!= null )

　　{

　　connection.close();

　　}

　　}

　　}

　　}　　　

开始点是连接工厂查找，这个工厂用于创建一个连接：

factory= jndiUtil.getQueueConnectionFactory("TestQM_QCF");

connection = factory.createQueueConnection();　　　

连接对象用于创建一个会话：

　session = connection.createQueueSession( transacted, Session.AUTO_ACKNOWLEDGE);　　

要将消息写入IN.QUEUE queue，查找前面创建的目的地对象OutputTestQueue：

oQueue= jndiUtil.getQueue("OutputTestQueue");　　

最后创建一个QueueSender对象将消息写入队列：

queueSender = session.createSender(oQueue);
TextMessage oMsg = session.createTextMessage();
oMsg.setText("www.devx.com");
queueSender.send(oMsg);　　

从OUT.QUEUE读取消息的过程相同，但使用的是QueueReceiver。
编译运行示例类

　　当你安装WMQ时会自动将编译和运行示例类需要的jar文件添加到CLASSPATH环境变量中，需要的jar文件位于C:\Program Files\IBM\WebSphere MQ\Java\lib，包括JMS和JNDI需要的jar。

　　在运行Tester类之前，使用MQ管理器向IN.QUEUE增加一条测试消息：

　　1.在 IN.QUEUE上点击右键，选择放入测试消息。

　　2.输入任意文本，点击放入消息。

　　如果你还没有在IN.QUEUE队列中放入过消息，Tester类会显示“队列中无消息”。

　　要运行Tester类，将Tester.class 和 JNDIUtil.class添加到CLASSPATH，然后在命令提示符输入：

　java devx.articles.mqjms.Tester　　　应用程序向OUT.QUEUE写入一条消息，并显示从IN.QUEUE检索到的消息，要检查发送到OUT.QUEUE中的消息，进入MQ管理器，在OUT.QUEUE上点击右键?选择“浏览消息”即可。

　　企业中的Java和WMQ MOM

　　在大型企业环境下，可以充分利用WMQ的性能和稳定性优势，只要你的代码遵循Java标准接口，你就可以受益。
 

本篇文章来源于：开发学院 http://edu.codepub.com   原文链接：http://edu.codepub.com/2009/0419/3122_2.php

 

 

这个是自己写的

 

所使用jar包为：ibmmq.jar,com.ibm.mq.jar,jms.jar,com.ibm.mq_jms.jar

 

 

 

package com.learn.mq;
import com.ibm.mq.*;
import javax.jms.*;

import com.ibm.mq.jms.*;
public class PTPSender {
 
 String host = "localhost";
 String qm = "learn.mq";
 //服务器连接通道，前面的为自己建的，后面的为系统默认的
 String channel = "LEARN.SVRCON.CHANNEL";//"SYSTEM.DEF.SVRCONN";
 String queue = "LEARN.QUEUE";
 QueueConnection qConn = null;
 QueueSession session = null;
 //由此值得到回复消息的CorrelationID
 String msgID = null;
 public void init(){
  //连接工厂，用com.ibm.mq.jms中的类实现javax.jms中的接口
  QueueConnectionFactory qcf = new MQQueueConnectionFactory();
  
  //设置连接工厂属性
  try {
   ((MQQueueConnectionFactory)qcf).setHostName(host);
   ((MQQueueConnectionFactory)qcf).setQueueManager(qm);
   ((MQQueueConnectionFactory)qcf).setCCSID(1381);
   ((MQQueueConnectionFactory)qcf).setChannel(channel);
   ((MQQueueConnectionFactory)qcf).setPort(11001);
   qConn = qcf.createQueueConnection();
   
   //JMS规范指出要在停止的状态下创建连接，使用时需要start
   qConn.start();
  } catch (JMSException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
   return;
  }
  
  System.out.println("MQ连接成功");
  
 }
 //将消息发送到inputQ中
 public void sendMessage(){
  try{
   boolean transacted = false;
   
   //JMS会话是创建和使用消息的单线程上下文
   //非事务处理（分别接收或发送消息）[事务处理（全部发送或者全部接收作为一个单元的一组消息）]
   session = qConn.createQueueSession(transacted
     , Session.AUTO_ACKNOWLEDGE);
   
   //对队列管理器上队列的映射
   Queue inputQ = session.createQueue("inputQ");
   QueueSender qSender = session.createSender(inputQ);
   
   //消息由会话创建
   TextMessage message = session.createTextMessage();
   message.setText("this is input message from Sender");
   
   //消息为非持久消息
   message.setJMSDeliveryMode(DeliveryMode.NON_PERSISTENT);
   
   //创建队列的另一种方式，learn.mq也可以指定为其它队列管理器
   Queue outputQ = session.createQueue("queue://learn.mq/outputQ");
   //设置消息的回复队列，JMSReplyTo为消息头
   message.setJMSReplyTo(outputQ);
   
   //发送消息，里面后两个参数暂时不知道什么含义
   qSender.send(message,DeliveryMode.NON_PERSISTENT,7,0);
   
   //获得发送消息的ID，放入msgID中暂存，留给取消息时用
   String messageID = message.getJMSMessageID();
   msgID = messageID;
   
  } catch (JMSException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
 
 //将消息从inputQ中转移到outputQ并设置消息的CorrectionID
 public void moveMessage(){
  Queue moveQ_get;
  Queue moveQ_put;
  try {
   moveQ_get = session.createQueue("inputQ");
   QueueReceiver qReceiver = session.createReceiver(moveQ_get);
   TextMessage message = (TextMessage)qReceiver.receive();
   
   message.setJMSCorrelationID(message.getJMSMessageID());
   
   System.out.println(message.getJMSReplyTo().toString());
   moveQ_put = session.createQueue(message.getJMSReplyTo().toString());
   QueueSender qSender = session.createSender(moveQ_put);
   qSender.send(message);
  } catch (JMSException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
  
 }
 
 public void receiveMessage(){
  try{
   Queue outputQ = session.createQueue("outputQ");
   
   //获取发送消息对应的回复消息
   QueueReceiver qReceiver = session.createReceiver(outputQ
     ,"JMSCorrelationID='"+msgID+"'");
   //设置接收消息的超时时间
   TextMessage message = (TextMessage)qReceiver.receive(10000);
   System.out.println("receive message is:"+message.getText());
   
   //注册的监听器需要在一线程中运行
   //MyMsgListener msgListener = new MyMsgListener();
   //qReceiver.setMessageListener(msgListener);
   //wait();
   
  } catch (JMSException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
 
 //释放会话及连接
 public void destroy(){
  
  try {
   session.close();
   qConn.close();
  } catch (JMSException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
 
 
 public static void main(String[] args) {
  PTPSender sender = new PTPSender();
  sender.init();
  sender.sendMessage();
  sender.moveMessage();
  sender.receiveMessage();
  sender.destroy();
 }

}

//消息监听器
class MyMsgListener implements MessageListener{
 public void onMessage(Message message){
  try{
   if(message instanceof TextMessage){
    System.out.println("Listener 接收消息："+((TextMessage)message).getText());
   }
  }catch(JMSException e){
   e.printStackTrace();
  }
 }
}

//异常监听器,需要有消息监听器一同运行，它将会发送MyExceptionListener对象到onMessage中
//异常监听器注册到连接对象上，也就是qConn.setExceptionListener(myExceptionListener)
class MyExceptionListener implements ExceptionListener{

 //此处可放入更多逻辑
 public void onException(JMSException e) {
  System.out.println("队列管理器异常");
  e.printStackTrace();
  System.exit(0);
 }
 
}