宁静海

该Sample为用JMS直接访问MQ的简单例子，其中牵涉3个最基本的JMS模型，Producer/Consumer/Browser，因为在该Sample中Browser作用更像是个Monitor，所以类名使用的是Monitor。

注意一点，该Sample连接方式使用的是Bind方式，该方式将会自动连接本地的MQ服务器，如果改成Client，除非连接远程客户机，并且有正确的用户设置方式，不然即使host写成本地IP，依然会抛 2035 (MQRC_NOT_AUTHORIZED) 错误。

Prducer

package jms.direct;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

import com.ibm.msg.client.jms.JmsConnectionFactory;
import com.ibm.msg.client.jms.JmsFactoryFactory;
import com.ibm.msg.client.wmq.WMQConstants;

public class JMSProducer {
	
	private String MQ_HOST = "localhost";
	private int MQ_PORT = 1416;
	private String MQ_CHANNEL = "";
	private String MQ_QUEUEMANAGER_NAME = "MW_QM";
	private String MQ_QUEUE_NAME = "MW_QUEUE";

	private JmsFactoryFactory jff = null;
	private JmsConnectionFactory jcf = null;
	private Connection connection = null;
	private Session session = null;
	private Queue queue = null;
	private MessageProducer producer = null;
	private Message msg = null;

	public JMSProducer() {
		try {
			// Create a connection factory
			jff = JmsFactoryFactory.getInstance(WMQConstants.WMQ_PROVIDER);
			jcf = jff.createConnectionFactory();
			// Set the properties
//			jcf.setStringProperty(WMQConstants.WMQ_HOST_NAME, MQ_HOST);
//			jcf.setIntProperty(WMQConstants.WMQ_PORT, MQ_PORT);
//			jcf.setStringProperty(WMQConstants.WMQ_CHANNEL, MQ_CHANNEL);
//			jcf.setStringProperty(WMQConstants.USERID, "");
//			jcf.setStringProperty(WMQConstants.PASSWORD, "");
//			jcf.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_CLIENT);
			jcf.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_BINDINGS);
			jcf.setStringProperty(WMQConstants.WMQ_QUEUE_MANAGER, MQ_QUEUEMANAGER_NAME);

			// Create JMS objects
			connection = jcf.createConnection();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

			queue = session.createQueue(MQ_QUEUE_NAME);
			producer = session.createProducer(queue);
			
			connection.start();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}
	
	public void sendMsg(String msgStr, String msgId) {
		try {
			msg = session.createMessage();
			msg.setStringProperty("STRMSG", msgStr);
			msg.setJMSCorrelationID(msgId);
			producer.send(msg);
		} catch (JMSException e) {
			e.printStackTrace();
		}
		System.out.println("-------------");
		System.out.println("Message send complete!");
		System.out.println("MessageId: " + msgId);
		System.out.println("Message: " + msgStr);
	}
	
	public void close() throws JMSException {
		producer.close();
		session.close();
		connection.close();
    }

	public static void main(String[] args) {
		try {
			System.out.println("Starting ...");
			JMSProducer producer = new JMSProducer();
			producer.sendMsg("Hello1", "000000001");
			producer.sendMsg("Hello2", "000000002");
			producer.sendMsg("Hello3", "000000003");
			producer.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}

Consumer

package jms.direct;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;

import com.ibm.msg.client.jms.JmsConnectionFactory;
import com.ibm.msg.client.jms.JmsFactoryFactory;
import com.ibm.msg.client.wmq.WMQConstants;

public class JMSConsumer {
	private String MQ_QUEUEMANAGER_NAME = "MW_QM";
	private String MQ_QUEUE_NAME = "MW_QUEUE";

	private JmsFactoryFactory jff = null;
	private JmsConnectionFactory jcf = null;
	private Connection connection = null;
	private Session session = null;
	private Queue queue = null;
	private MessageConsumer consumer = null;
	private Message msg = null;

	public JMSConsumer() {
		try {
			// Create a connection factory
			jff = JmsFactoryFactory.getInstance(WMQConstants.WMQ_PROVIDER);
			jcf = jff.createConnectionFactory();
			// Set the properties
			jcf.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_BINDINGS);
			jcf.setStringProperty(WMQConstants.WMQ_QUEUE_MANAGER, MQ_QUEUEMANAGER_NAME);

			// Create JMS objects
			connection = jcf.createConnection();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

			queue = session.createQueue(MQ_QUEUE_NAME);
			consumer = session.createConsumer(queue);
			
			connection.start();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}
	
	public void consume() {
		try {
			msg = consumer.receive();
			
			String msgId = msg.getJMSCorrelationID();
			String msgStr = msg.getStringProperty("STRMSG");
			
			System.out.println("-------------");
			System.out.println("Message received!");
			System.out.println("MessageId: " + msgId);
			System.out.println("Message: " + msgStr);
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}
	
	public void close() throws JMSException {
		consumer.close();
		session.close();
		connection.close();
    }
	
	public static void main(String[] args) {
		try {
			System.out.println("Receiving ...");
			JMSConsumer receiver = new JMSConsumer();
			receiver.consume();
			receiver.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

Browser

package jms.direct;

import java.util.ArrayList;

import java.util.Enumeration;

import java.util.List;
import javax.jms.Connection;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.Queue;

import javax.jms.QueueBrowser;

import javax.jms.Session;
import com.ibm.msg.client.jms.JmsConnectionFactory;

import com.ibm.msg.client.jms.JmsFactoryFactory;

import com.ibm.msg.client.wmq.WMQConstants;
public class JMSMonitor {

    private String MQ_QUEUEMANAGER_NAME = "MW_QM";

    private String MQ_QUEUE_NAME = "MW_QUEUE";
    private JmsFactoryFactory jff = null;

    private JmsConnectionFactory jcf = null;

    private Connection connection = null;

    private Session session = null;

    private Queue queue = null;

    private QueueBrowser monitor = null;

    private Message msg = null;

    private List<string> msgIdList = null; 
    public void initMsgIdList() {

        msgIdList = new ArrayList<string>();

        msgIdList.add("000000001");

        msgIdList.add("000000002");

        msgIdList.add("000000003");     }      public JMSMonitor() {

        initMsgIdList();

        try {

            // Create a connection factory

            jff = JmsFactoryFactory.getInstance(WMQConstants.WMQ_PROVIDER);

            jcf = jff.createConnectionFactory();

            // Set the properties

            jcf.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_BINDINGS);

            jcf.setStringProperty(WMQConstants.WMQ_QUEUE_MANAGER, MQ_QUEUEMANAGER_NAME);

            // Create JMS objects

            connection = jcf.createConnection();

            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            queue = session.createQueue(MQ_QUEUE_NAME);

            monitor = session.createBrowser(queue);

            connection.start();

        } catch (JMSException e) {

            e.printStackTrace();

        }

    }
    public void startMonitor() {

        while (true) {

            if (msgIdList.size() > 0) {

                try {

                    Enumeration<message> msgs = monitor.getEnumeration();

                    List<string> remainMsgIdList = new ArrayList<string>();

                    while (msgs.hasMoreElements()) {

                        Message msg = msgs.nextElement();

                        String msgId = msg.getJMSCorrelationID();

                        if (msgIdList.contains(msgId)) {

                            remainMsgIdList.add(msgId);

                            System.out.println("Message ID [" + msgId + "] found, this message still not be consumed!");

                        } else if (!msgIdList.contains(msgId)) {

                            System.out.println("Message ID [" + msgId + "] is not be recorded by the sender yet, just ignnore it.");

                        }

                    }

                    msgIdList = remainMsgIdList;

                    // All message scan complete.

                    if (msgIdList.size() > 0) {

                        System.out.println("------------------------------");

                        System.out.println("Following Msg not be consumed!");

                        System.out.println("------------------------------");

                        for (String msgId : msgIdList) {

                            System.out.println(msgId);

                        }

                    } else {

                        System.out.println("No message record in the data base.");

                        // The sample end here, for the list will not be added

                        // again.

                    }

                } catch (JMSException e) {

                    e.printStackTrace();

                }

            }

            synchronized (this) {

                try {

                    this.wait(5000);

                } catch (InterruptedException ie) {

                    ie.printStackTrace();

                }

            }