一段同步接收和发送MQ消息的代码

时间：2015-10-14 00:29:09 阅读：346 评论：0 收藏：0 [点我收藏+]

标签：

JAVA代码:

Java代码

package com.sdb.payment.core.mq;
import org.apache.log4j.Logger;
import com.ibm.mq.MQC;
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;
public class MessageQueueService {
private static Logger logger = Logger.getLogger(MessageQueueService.class);
private String hostname = "192.168.0.117";
private String channel = "CHL.SVRCONN";
private String queueManager = "QM_SERVER";
private String sendQueue = "OMP.QRMT";
private String recvQueue = "OMP.QLCA";
private int port = 24100;
private int ccsid = 1381;
private int failedCount = 5;
private int intervalTime = 1000;
public MessageQueueService() {
MQEnvironment.hostname = hostname;
MQEnvironment.channel = channel;
MQEnvironment.CCSID = ccsid;
MQEnvironment.port = port;
}
public String send(String sendMsg) throws Exception {
MQQueueManager qManager = new MQQueueManager(queueManager);
// send message
int openOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING;
MQQueue sQueue = qManager.accessQueue(sendQueue, openOptions);
MQPutMessageOptions pmo = new MQPutMessageOptions();
MQMessage send = new MQMessage();
send.write(sendMsg.getBytes());
System.out.println("send message : " + sendMsg);
sQueue.put(send, pmo);
sQueue.close();
System.out.println("send message Id");
for (int i = 0; i<send.messageId.length; i++) {
System.out.print(send.messageId[i]);
}
System.out.println();
System.out.println("send message Id");
// fetch message
openOptions = MQC.MQOO_INQUIRE + MQC.MQOO_FAIL_IF_QUIESCING
+ MQC.MQOO_INPUT_SHARED;
MQQueue rQueue = qManager.accessQueue(recvQueue, openOptions);
MQGetMessageOptions getOptions = new MQGetMessageOptions();
getOptions.options = MQC.MQGMO_WAIT;
getOptions.waitInterval = intervalTime;
MQMessage recvMsg = new MQMessage();
recvMsg.messageId = send.messageId;//这里是关键，要保持接收的msgid跟发送的msgid值是一样的，
//这样就会根据msgId来取队列的消息了，而不会取到别的消息
send.clearMessage();
boolean received = false;
int fetchCount = 0;
while (!received) {
try {
fetchCount++;
rQueue.get(recvMsg, getOptions);
//logger.debug("the " + fetchCount + " time fetch message!");
System.out.println("fetch message !!!");
received = true;
} catch (MQException me) {
if (me.reasonCode == MQException.MQRC_NO_MSG_AVAILABLE) {
if (fetchCount > failedCount) {
recvMsg.clearMessage();
rQueue.close();
qManager.disconnect();
//logger.error("can‘t fetch message for " + me.getMessage());
return null;
}
}
} catch (Exception ex) {
recvMsg.clearMessage();
rQueue.close();
qManager.disconnect();
//logger.error("can‘t fetch message for " + ex.getMessage());
return null;
}
}
byte[] bMsg = new byte[recvMsg.getMessageLength()];
recvMsg.readFully(bMsg);
System.out.println("rec correlationId Id");
for (int i = 0; i<recvMsg.correlationId.length; i++) {
System.out.print(recvMsg.correlationId[i]);
}
System.out.println();
System.out.println("rec correlationId Id");
String recv = new String(bMsg);
recvMsg.clearMessage();
rQueue.close();
qManager.disconnect();
return recv;
}
public void setChannel(String channel) {
this.channel = channel;
}
public void setHostname(String hostname) {
this.hostname = hostname;
}
public void setQueueManager(String queueManager) {
this.queueManager = queueManager;
}
public void setPort(int port) {
this.port = port;
}
public void setIntervalTime(int intervalTime) {
this.intervalTime = intervalTime;
}
public void setFailedCount(int failedCount) {
this.failedCount = failedCount;
}
public void setRecvQueue(String recvQueue) {
this.recvQueue = recvQueue;
}
public void setSendQueue(String sendQueue) {
this.sendQueue = sendQueue;
}
}

一段同步接收和发送MQ消息的代码

标签：

原文地址：http://my.oschina.net/u/2273085/blog/516843

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行