使用JMS实现MQ消息处理

MQ针对Java语言提供了两种接口来实现消息的发送和接收，这两种接口分别是：

Ø         Java

Ø         JMS

在北京地税财税库行横向联网项目中采用了MQ作为消息中间件，消息的发送和接收采用的是Java接口，在本文中将说明如何采用JMS实现MQ消息发送和接收功能。

2       JMS实现 MQ消息处理技术说明

如果使用JMS来访问MQ，则应当使用一个目录服务器来实现MQ连接工厂和队列资源的注册，以供客户端查找，客户端的在获取JMS资源的应选择LDAP。





3       环境准备
本次测试采用如下软件进行测试

Ø         MQ 6.0 for windows

Ø         IBM Tivoli Directory Server 6.0 （LDAP）

Ø         WebLogic 10.1

4       创建MQ对象
进入到MQ队列管理器创建中创建队列管理器和队列

1）  创建队列管理器QM_YFZX

2）  在队列管理器中创建队列QUEUE_TEST

5       LDAP配置
在TDS中创建一个LDAP分支dc=liutg用于MQ JMS注册使用，其访问的URL为：ldap://192.100.8.244:389/dc=liutg。

使用MQ提供的JMSAdmin工具实现连接工厂和队列的注册，步骤如下

1)      进入到%MQ%\Java\bin目录

2)      编辑JMSAdmin.config文件，按照下面内容进行修改

INITIAL_CONTEXT_FACTORY=com.sun.jndi.ldap.LdapCtxFactory

PROVIDER_URL=ldap://192.100.8.244:389/dc=liutg

SECURITY_AUTHENTICATION=simple

PROVIDER_USERDN=cn=root

PROVIDER_PASSWORD=admin123

3)      运行JMSAdmin命令





注册队列连接工厂

Def qcf(ConnectionFactory-mq) qmgr(QM_YFZX)





注册队列

Def q(Queue-mq)  qmgr(QM_YFZX) queue(QUEUE_TEST)









查看注册上下文：

Dis ctx





查看队列连接工厂

Dis qcf(cn= ConnectionFactory-mq)





查看队列

Dis q(cn=Queue-mq)






6       使用JMS发送消息
6.1    处理
1）获取上下文

    Hashtable<String,String> env = new Hashtable<String,String>();

    env.put(Context.INITIAL_CONTEXT_FACTORY, “com.sun.jndi.ldap.LdapCtxFactory”);

    env.put(Context.PROVIDER_URL, “ldap://192.100.8.244:389/dc=liutg”);

Context ctx = new InitialContext(env);

QueueConnectionFactory qconFactory =

(QueueConnectionFactory) ctx.lookup(“cn=ConnectionFactory-mq”);

       QueueConnection qcon = qconFactory.createQueueConnection();

    QueueSession qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

    Queue queue = (Queue) ctx.lookup(queueName);

    QueueSender qsender = qsession.createSender(queue);


    TextMessage msg = qsession.createTextMessage();

qcon.start();

msg.setText(“hello …”);

       qsender.send(msg);

      

       qsender.close();

    qsession.close();

    qcon.close();













6.2    示例
public class LdapQueueSend

{

  // Defines the JNDI context factory. LDAP

  public final static String JNDI_FACTORY="com.sun.jndi.ldap.LdapCtxFactory";






  // Defines the JMS context factory.

  public final static String JMS_FACTORY="cn=ConnectionFactory-mq";






  // Defines the queue.

  public final static String QUEUE="cn=Queue-mq";






  private QueueConnectionFactory qconFactory;

  private QueueConnection qcon;

  private QueueSession qsession;

  private QueueSender qsender;

  private Queue queue;

  private TextMessage msg;





  /**

   * Creates all the necessary objects for sending

   * messages to a JMS queue.

   *

   * @param ctx JNDI initial context

   * @param queueName name of queue

   * @exception NamingException if operation cannot be performed

   * @exception JMSException if JMS fails to initialize due to internal error

   */

  public void init(Context ctx, String queueName)

    throws NamingException, JMSException

  {

    qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);

    qcon = qconFactory.createQueueConnection();

    qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

    queue = (Queue) ctx.lookup(queueName);

    qsender = qsession.createSender(queue);

    msg = qsession.createTextMessage();

    qcon.start();

  }





  /**

   * Sends a message to a JMS queue.

   *

   * @param message  message to be sent

   * @exception JMSException if JMS fails to send message due to internal error

   */

  public void send(String message) throws JMSException {

    msg.setText(message);

    msg.setJMSCorrelationID("Req");

    qsender.send(msg);

  }





  /**

   * Closes JMS objects.

   * @exception JMSException if JMS fails to close objects due to internal error

   */

  public void close() throws JMSException {

    qsender.close();

    qsession.close();

    qcon.close();

  }

/** main() method.

  *

  * @param args WebLogic Server URL

  * @exception Exception if operation fails

  */

  public static void main(String[] args) throws Exception {

       //weblogic 10  FILE

       InitialContext ic = getInitialContext("ldap://192.100.8.244:389/dc=liutg");


    LdapQueueSend qs = new LdapQueueSend();

    qs.init(ic, QUEUE);

    readAndSend(qs);

    qs.close();

  }





  private static void readAndSend(LdapQueueSend qs)

    throws IOException, JMSException

  {

    BufferedReader msgStream = new BufferedReader(new InputStreamReader(System.in));

    String line=null;

    boolean quitNow = false;

    do {

      System.out.print("Enter message (\"quit\" to quit): \n");

      line = msgStream.readLine();

      if (line != null && line.trim().length() != 0) {

        qs.send(line);

        System.out.println("JMS Message Sent: "+line+"\n");

        quitNow = line.equalsIgnoreCase("quit");

      }

    } while (! quitNow);





  }





  private static InitialContext getInitialContext(String url)

    throws NamingException

  {

    Hashtable<String,String> env = new Hashtable<String,String>();

    env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);

    env.put(Context.PROVIDER_URL, url);

    return new InitialContext(env);

  }





}





7       使用JMS接收消息




package com.syax.study.jms;





import java.util.Hashtable;

import javax.jms.*;

import javax.naming.Context;

import javax.naming.InitialContext;

import javax.naming.NamingException;

import javax.transaction.UserTransaction;





/**

* This example shows how to establish a connection to

* and receive messages from a JMS queue in a client-demarcated

* transaction.  The classes in this package operate on the same

* JMS queue. Run the classes together to witness messages being

* sent and received, and to browse the queue for messages.

*

* @author Copyright (c) 1999-2006 by BEA Systems, Inc. All Rights Reserved.

*/

public class LdapQueueReceiveInTx

{

// Defines the JNDI context factory.

         public final static String JNDI_FACTORY="com.sun.jndi.ldap.LdapCtxFactory";






  // Defines the JMS connection factory for the queue.

  public final static String JMS_FACTORY="cn=ConnectionFactory-mq";






  // Defines the queue.

  public final static String QUEUE="cn=Queue-mq";





  private QueueConnectionFactory qconFactory;

  private QueueConnection qcon;

  private QueueSession qsession;

  private QueueReceiver qreceiver;

  private Queue queue;

//  private UserTransaction utx;





/**

  * Receives message interface.

  */

  public void receiveMessages() throws Exception {

    Message msg = null;

    String msgText = "";





    try {

      // Set transaction timeout to 30 minutes.

//      utx.setTransactionTimeout(1800);

//      utx.begin();

      System.out.println("TRANSACTION BEGUN");

      do {

        msg = qreceiver.receive();

      

        if (msg != null) {

          if (msg instanceof TextMessage) {

            msgText = ((TextMessage)msg).getText();

          } else {

            msgText = msg.toString();

          }

          System.out.println("Message Received: "+ msgText );

          if (msgText.equalsIgnoreCase("rollback")) {

//              utx.rollback();

              System.out.println("TRANSACTION RollBacked");

            }

          if (msgText.equalsIgnoreCase("quit")) {

//            utx.commit();

            System.out.println("TRANSACTION COMMITTED");

          }

        }

      } while(msg != null && ! msgText.equalsIgnoreCase("quit"));

    } catch (JMSException jmse) {

      System.out.println("Error receiving JMS message: "+jmse);

      System.err.println("An exception occurred: "+jmse.getMessage());

      throw jmse;

    }

  }





  /**

   * Creates all the necessary objects for receiving

   * messages from a JMS queue.

   *

   * @param   ctx     JNDI initial context

   * @param queueName     name of queue

   * @exception NamingException operation cannot be performed

   * @exception JMSException if JMS fails to initialize due to internal error

   */

  public void init(Context ctx, String queueName)

    throws NamingException, JMSException

  {

    qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);

    qcon = qconFactory.createQueueConnection();

    qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

    //utx = (UserTransaction) ctx.lookup("javax.transaction.UserTransaction");

    queue = (Queue) ctx.lookup(queueName);

   

//    qreceiver = qsession.createReceiver(queue);

    qreceiver = qsession.createReceiver(queue,"JMSCorrelationID='Req'");

    qcon.start();

  }





  /**

   * Closes JMS objects.

   * @exception JMSException if JMS fails to close objects due to internal error

   */

  public void close() throws JMSException {

    qreceiver.close();

    qsession.close();

    qcon.close();

  }





/**

  * main() method.

  *

  * @param args  WebLogic Server URL

  * @exception  Exception if execution fails

  */

  public static void main(String[] args) throws Exception {





    InitialContext ic = getInitialContext("ldap://192.100.8.244:389/dc=liutg");


    LdapQueueReceiveInTx qr = new LdapQueueReceiveInTx();

    qr.init(ic, QUEUE);





    System.out.println("JMS Ready To Receive Messages (To quit, send a \"quit\" message).");





    qr.receiveMessages();

    qr.close();

  }





  private static InitialContext getInitialContext(String url)

    throws NamingException

  {

    Hashtable<String,String> env = new Hashtable<String,String>();

    env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);

    env.put(Context.PROVIDER_URL, url);

    return new InitialContext(env);

  }





}

















8       与WebLogic集成
WebLogic与MQ集成有两种方法，一种是通过JMS Bridge实现，这种方法是将消息从MQ队列中放入到WebLogic Server中的队列，由于MQ优于WebLogic server自带的队列，这种方法MQ不能有效发挥其作用并且额外增加了WebLogic Server的处理量，因此不做考虑；第二种方法为创建一个外部的JMS Server，具体方法如下，

1）  设置WebLogic ClassPath。将MQ的java类包拷贝到WebLogic Domain的Lib目录下，启动WebLogic，进入WebLogic控制台

2）  创建一个JMS Module

3）  在JMS Module中创建一个Foreign Server

JNDI Initial Context Factory:  com.sun.jndi.ldap.LdapCtxFactory

JNDI Connection URL    ldap://192.100.8.244:389/dc=liutg

4）  在新创建的Foreign Server中

创建一个ForeignDestination，

Local JNDI Name：Queue-mq


Remote JNDI Name：cn=Queue-mq


创建一个ConnectionFactory

Local JNDI Name：ConnectionFactory-mq


Remote JNDI Name：cn= ConnectionFactory-mq


       

       这样，WebLogic就可以使用本地定义的ConnectionFactory和Queue来访问MQ。发送消息采用JMS发送，接收消息可以采用MDB实现。