java与websphere mq的交互_JAVA_编程开发_程序员俱乐部

Test Team 一直用 rfhutil 对MQ项目进行测试，而 rfhutil 每次只能对一个 Queue 进行读写操作，一旦有批量的 Queue 需要进行测试时，工作变得特别繁琐。
领导找到我们几个同事，用 java 做了一个测试 MQ 的小工具。
这个测试工具的主要工作流程为：将本地的用作测试的请求数据，发送至 Input MQ，经过MQ项目的内部处理之后，接收 Output MQ 的数据，将这个数据与本地的期望数据进行比对，如果结果为true，则表示测试通过，如果为false，则表示测试失败。另外还有一些零碎的需求，如清空 Output MQ、保存 Output MQ 的数据等。
这里主要讨论该工具中java与mq的交互方式。

1、首先将mq的jar包准备好；



2、1）使用com.ibm.mq.jar下的 MQEnvironment、MQQueueManager、MQQueue等工具类，与MQ建立连接；
   2）建立连接之后，使用 MQMessage、MQGetMessageOptions、MQPutMessageOptions等进行发送和接收数据。
连接类源码如下：
import com.ibm.mq.MQEnvironment;
import com.ibm.mq.MQException;
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;
import com.ibm.mq.constants.CMQC;

public class MQConnection {

private MQQueueManager mqQueueManager;
private MQQueue defaultQueue;

//建立连接
public MQConnection(String hostName, String channel, String queueManagerName, String queue, int portNumber)

{
MQEnvironment.hostname = hostName;
MQEnvironment.channel = channel;
MQEnvironment.port = portNumber;
int openOptions = CMQC.MQOO_INPUT_AS_Q_DEF | CMQC.MQOO_OUTPUT | CMQC.MQOO_INQUIRE;
try {
mqQueueManager = new MQQueueManager(queueManagerName);
defaultQueue = mqQueueManager.accessQueue(queue, openOptions, null, queue, null);
} catch (MQException e) {
e.printStackTrace();
}
}

//发送数据至MQ
public void placeMessageOnQueue(MQMessage msg, MQPutMessageOptions options)
throws MQException {
defaultQueue.put(msg, options);
}

//从MQ中获取数据
public MQMessage getMessageFromQueue() {
MQMessage msg = new MQMessage();
try {
MQGetMessageOptions mqgetmessageoptions = new MQGetMessageOptions();
mqgetmessageoptions.waitInterval = 0x7a120;
defaultQueue.get(msg, mqgetmessageoptions);
} catch (MQException ex) {
System.out.println("An MQ error occurred : Completion code "
+ ex.completionCode + " Reason code " + ex.reasonCode);
ex.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
}
return msg;
}

//获取MQ的深度
public int getCurrentDepth() {
try {
return defaultQueue.getCurrentDepth();
} catch (MQException e) {
e.printStackTrace();
}
return 0;
}

//使用示例：
public static void main(String args[]) {

private String inputQueueHost = "此处为有效的IP";
private String inputQueueChannel = "此处为有效的Channel信息";
private String inputQueueManager = "此处为有效的QManager名称";
private String inputQueueName = "此处为有效的QueueName";
private int inputQueuePortNumber = [此处为有效的MQ端口];

FileInputStream fileInputStream = null;
try {
MQConnection mqi = new MQConnection(inputQueueHost,
inputQueueChannel,
inputQueueManager,
inputQueueName,
inputQueuePortNumber);

String inputFilePath = "【此处为有效的文件路径】";
File inputFile = new File(inputFilePath);

fileInputStream = new FileInputStream(inputFile);
byte inputMessageContent[] = new byte[(int)inputFile.length()];
fileInputStream.read(inputMessageContent);
MQMessage message = new MQMessage();
message.setStringProperty("fileName", inputFile.getName());
message.write(inputMessageContent);
MQPutMessageOptions putMessageOptions = new MQPutMessageOptions();
mqi.placeMessageOnQueue(message, putMessageOptions);
} catch (Exception e) {
e.printStackTrace();
System.err.println(e);
} finally {
if(fileInputStream!=null) {
try {
fileInputStream.close();
} catch(Exception e) {
e.printStackTrace();
}
}
}
}
}

只可惜没有 com.ibm.mq 包的源码，无法了解具体的实现方式。

• 
• 大小: 80.1 KB

• 查看图片附件