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

JAVA代码:

Java代码  

1. package com.sdb.payment.core.mq;
2. import org.apache.log4j.Logger;
3. import com.ibm.mq.MQC;
4. import com.ibm.mq.MQEnvironment;
5. import com.ibm.mq.MQException;
6. import com.ibm.mq.MQGetMessageOptions;
7. import com.ibm.mq.MQMessage;
8. import com.ibm.mq.MQPutMessageOptions;
9. import com.ibm.mq.MQQueue;
10. import com.ibm.mq.MQQueueManager;
11. public class MessageQueueService {
12. private static Logger logger = Logger.getLogger(MessageQueueService.class);
13. private String hostname = "192.168.0.117";
14. private String channel = "CHL.SVRCONN";
15. private String queueManager = "QM_SERVER";
16. private String sendQueue = "OMP.QRMT";
17. private String recvQueue = "OMP.QLCA";
18. private int port = 24100;
19. private int ccsid = 1381;
20. private int failedCount = 5;
21. private int intervalTime = 1000;
22. public MessageQueueService() {
23. MQEnvironment.hostname = hostname;
24. MQEnvironment.channel = channel;
25. MQEnvironment.CCSID = ccsid;
26. MQEnvironment.port = port;
27. }
28. public String send(String sendMsg) throws Exception {
29. MQQueueManager qManager = new MQQueueManager(queueManager);
30. // send message
31. int openOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING;
32. MQQueue sQueue = qManager.accessQueue(sendQueue, openOptions);
33. MQPutMessageOptions pmo = new MQPutMessageOptions();
34. MQMessage send = new MQMessage();
35. send.write(sendMsg.getBytes());
36. System.out.println("send message : " + sendMsg);
37. sQueue.put(send, pmo);
38. sQueue.close();
39. System.out.println("send message Id");
40. for (int i = 0; i<send.messageId.length; i++) {
41. System.out.print(send.messageId[i]);
42. }
43. System.out.println();
44. System.out.println("send message Id");
45. // fetch message
46. openOptions = MQC.MQOO_INQUIRE + MQC.MQOO_FAIL_IF_QUIESCING
47. + MQC.MQOO_INPUT_SHARED;
48. MQQueue rQueue = qManager.accessQueue(recvQueue, openOptions);
49. MQGetMessageOptions getOptions = new MQGetMessageOptions();
50. getOptions.options = MQC.MQGMO_WAIT;
51. getOptions.waitInterval = intervalTime;
52. MQMessage recvMsg = new MQMessage();
53. recvMsg.messageId = send.messageId;//这里是关键，要保持接收的msgid跟发送的msgid值是一样的，
54. //这样就会根据msgId来取队列的消息了，而不会取到别的消息
55. send.clearMessage();
56. boolean received = false;
57. int fetchCount = 0;
58. while (!received) {
59. try {
60. fetchCount++;
61. rQueue.get(recvMsg, getOptions);
62. //logger.debug("the " + fetchCount + " time fetch message!");
63. System.out.println("fetch message !!!");
64. received = true;
65. } catch (MQException me) {
66. if (me.reasonCode == MQException.MQRC_NO_MSG_AVAILABLE) {
67. if (fetchCount > failedCount) {
68. recvMsg.clearMessage();
69. rQueue.close();
70. qManager.disconnect();
71. //logger.error("can‘t fetch message for " + me.getMessage());
72. return null;
73. }
74. }
75. } catch (Exception ex) {
76. recvMsg.clearMessage();
77. rQueue.close();
78. qManager.disconnect();
79. //logger.error("can‘t fetch message for " + ex.getMessage());
80. return null;
81. }
82. }
83. byte[] bMsg = new byte[recvMsg.getMessageLength()];
84. recvMsg.readFully(bMsg);
85. System.out.println("rec correlationId Id");
86. for (int i = 0; i<recvMsg.correlationId.length; i++) {
87. System.out.print(recvMsg.correlationId[i]);
88. }
89. System.out.println();
90. System.out.println("rec correlationId Id");
91. String recv = new String(bMsg);
92. recvMsg.clearMessage();
93. rQueue.close();
94. qManager.disconnect();
95. return recv;
96. }
97. public void setChannel(String channel) {
98. this.channel = channel;
99. }
100. public void setHostname(String hostname) {
101. this.hostname = hostname;
102. }
103. public void setQueueManager(String queueManager) {
104. this.queueManager = queueManager;
105. }
106. public void setPort(int port) {
107. this.port = port;
108. }
109. public void setIntervalTime(int intervalTime) {
110. this.intervalTime = intervalTime;
111. }
112. public void setFailedCount(int failedCount) {
113. this.failedCount = failedCount;
114. }
115. public void setRecvQueue(String recvQueue) {
116. this.recvQueue = recvQueue;
117. }
118. public void setSendQueue(String sendQueue) {
119. this.sendQueue = sendQueue;
120. }
121. }