JAVA 调用MQ

标签：

前提：MQ服务器端创建

 队列管理器：QueueManager

队列：XML.IN

服务器连接通道：SYSTEM.ADMIN.SVRCONN 

监听器监听1414端口。

package mq;

import com.ibm.mq.MQC;
import com.ibm.mq.MQEnvironment;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;

/**
 * 测试调用MQ
 * 
 * @author Administrator
 * 
 */
public class TestMQ {

	/**
	 * 队列管理器
	 */
	private static final String MQ_QUEUE_MANAGER = "QueueManager";

	/**
	 * 队列
	 */
	private static final String MQ_QUEUE = "XML.IN";

	/**
	 * 通道
	 */
	private static final String MQ_CHANNEL = "SYSTEM.ADMIN.SVRCONN";

	/**
	 * IP
	 */
	private static final String MQ_HOST = "localhost";

	/**
	 * 端口
	 */
	private static final int MQ_PORT = 1414;
	

	private MQQueueManager qMgr;
	
	/**
	 * 测试
	 * @param args
	 */
	public static void main(String[] args) {
		TestMQ test = new TestMQ();
		test.getConnMQManager();
		test.sendMessage("1111111");
		test.receiveMsg();
	}

	/**
	 * 发送消息到MQ
	 * @param message
	 */
	public void sendMessage(String message) {
		MQQueue mqQueue = null;
		try {
			// 建立连接
			int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT| MQC.MQOO_INQUIRE;
			mqQueue = qMgr.accessQueue(MQ_QUEUE, openOptions, null, null, null);
			MQMessage msg = new MQMessage();
			msg.format = MQC.MQFMT_STRING;
			msg.characterSet = 1381;
			msg.writeObject(message);
			MQPutMessageOptions pmo = new MQPutMessageOptions();
			// 设置消息用不过期
			msg.expiry = -1; 
			// 将消息放入队列
			mqQueue.put(msg, pmo);
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		} finally{
			if(mqQueue!=null){
				try {
					mqQueue.close();
				} catch (Exception e2) {
					// TODO: handle exception
					e2.printStackTrace();
				}
			}
		}
	}
	
	/**
	 * 读取MQ消息
	 */
	public void receiveMsg(){
		MQQueue mqQueue = null;
		try {
			// 建立连接
			int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT| MQC.MQOO_INQUIRE;
			mqQueue = qMgr.accessQueue(MQ_QUEUE, openOptions, null, null, null);
			// 队列深度
			int depth = mqQueue.getCurrentDepth();
			System.out.println("当前队列深度为："+depth);
			while (depth-->0){
				// 要读取的消息
				MQMessage msg = new MQMessage();
				MQGetMessageOptions mqGet = new MQGetMessageOptions();
				mqQueue.get(msg, mqGet);
				System.out.println("消息的大小为:"+msg.getDataLength());
				System.out.println("消息的内容为："+msg.readObject());
			}
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		} finally{
			if(mqQueue!=null){
				try {
					mqQueue.close();
				} catch (Exception e2) {
					// TODO: handle exception
					e2.printStackTrace();
				}
			}
		}
	}

	/**
	 * 获取连接
	 */
	public void getConnMQManager() {
		MQEnvironment.hostname = MQ_HOST;
		MQEnvironment.channel = MQ_CHANNEL;
		MQEnvironment.CCSID = 1381; // 字符编码
		MQEnvironment.port = MQ_PORT;
		try {
			qMgr = new MQQueueManager(MQ_QUEUE_MANAGER);
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		}
	}

	/**
	 * 关闭连接
	 */
	public void closeConnMQManager() {
		if (qMgr != null) {
			try {
				qMgr.close();
			} catch (Exception e) {
				// TODO: handle exception
				e.printStackTrace();
			}
		}
	}
}

JAVA 调用MQ

标签：

原文地址：http://blog.csdn.net/xiang520jl/article/details/46547369