Java调用MQ队列

时间：2023-03-09 02:53:50
Java调用MQ队列

IBM MQ 6.0中设置两个队列，（远程队列、通道之类都不设置）。

队列管理器是XIR_QM_1502

队列名称是ESBREQ

IP地址是10.23.117.134（远程的一台电脑，跟我的电脑不在一个局域网内）

端口1414

CCSID 1208

MQ配置可以参考这个，有配图http://wenku.baidu.com/view/06d108d0360cba1aa811daa3.html

程序如下，发送线程两个，接收线程一个。接收完毕后就结束。

  1. /*
  2. * 创建日期 2012-7-10
  3. *
  4. * TODO 要更改此生成的文件的模板，请转至
  5. * 窗口 － 首选项 － Java － 代码样式 － 代码模板
  6. */
  7. package yerasel;
  8. /**
  9. * @author Fenglb E-mail:56553655@163.com
  10. * @version 创建时间：2009-4-30 下午04:13:38 类说明
  11. */
  12. import java.io.IOException;
  13. import com.ibm.mq.MQC;
  14. import com.ibm.mq.MQEnvironment;
  15. import com.ibm.mq.MQException;
  16. import com.ibm.mq.MQGetMessageOptions;
  17. import com.ibm.mq.MQMessage;
  18. import com.ibm.mq.MQPutMessageOptions;
  19. import com.ibm.mq.MQQueue;
  20. import com.ibm.mq.MQQueueManager;
  21. interface SomeConstants {
  22. String qManager = "XIR_QM_1502";//"XIR_QM"; //QueueManager name
  23. String qName = "ESBREQ";// Queue Name
  24. String strIP = "10.23.117.134";//"10.24.28.139";//"10.24.28.102";
  25. int iPort = 1502;//1414;
  26. String strChl = "SYSTEM.DEF.SVRCONN";// Server-Connection Channel
  27. int iCCSID = 1208;
  28. }
  29. class Sender implements Runnable, SomeConstants {
  30. public void run() {
  31. sendMessage();
  32. }
  33. public void sendMessage() {
  34. String name = Thread.currentThread().getName();
  35. System.out.println("进入线程" + name);
  36. MQQueueManager qMgr = null;
  37. // configure connection parameters
  38. MQEnvironment.hostname = strIP;
  39. // Server name or IP
  40. MQEnvironment.port = iPort;
  41. MQEnvironment.channel = strChl;
  42. MQEnvironment.CCSID = iCCSID;
  43. // java程序连接mq的方式有两种，一是客户机方式，一是绑定方式，
  44. // 默认是客户机方式，当mq部署在本地的时候，就需要用绑定方式
  45. // 本机IP是10.24.28.139连接10.23.117.134的时候不需要下句
  46. //MQEnvironment.properties.put(MQC.TRANSPORT_PROPERTY,
  47. //MQC.TRANSPORT_MQSERIES_BINDINGS);
  48. // Create a connection to the QueueManager
  49. System.out.println(name + " Connecting to queue manager: " + qManager);
  50. try {
  51. qMgr = new MQQueueManager(qManager);
  52. // Set up the options on the queue we wish to open
  53. int openOptions = MQC.MQMT_REQUEST | MQC.MQPMO_NEW_MSG_ID
  54. | MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING
  55. | MQC.MQOO_INPUT_AS_Q_DEF;
  56. // Now specify the queue that we wish to open and the open options
  57. System.out.println(name + " Accessing queue: " + qName);
  58. MQQueue queue = qMgr.accessQueue(qName, openOptions);
  59. // Define a simple WebSphere MQ Message ...
  60. // Specify the default put message options
  61. MQPutMessageOptions pmo = new MQPutMessageOptions();
  62. // Put the message to the queue
  63. System.out.println(name + " Sending a message...");
  64. MQMessage msg = new MQMessage();
  65. msg.messageId = "MSGID".getBytes();
  66. msg.messageType = MQC.MQMT_REQUEST;
  67. msg.replyToQueueName = "ESBREQ";
  68. // 在此测试一下 mq 的传输次列
  69. for (int j = 1; j < 5; j++) {
  70. msg.messageSequenceNumber = j;
  71. // write some text in UTF8 format
  72. try {
  73. String str = "Salemetsizbe Yerasel";
  74. str = str + " " + j;
  75. msg.writeUTF(str);
  76. queue.put(msg, pmo);
  77. msg.clearMessage();
  78. System.out.println(name + " putting the message... " + j);
  79. } catch (MQException mqe) {
  80. mqe.printStackTrace();
  81. break;
  82. } catch (IOException e1) {
  83. e1.printStackTrace();
  84. }
  85. }
  86. qMgr.commit();
  87. System.out.println(name + " Done!");
  88. System.out.println("==========");
  89. System.out.println("");
  90. } catch (MQException e) {
  91. e.printStackTrace();
  92. }
  93. }
  94. }
  95. class Receiver implements Runnable, SomeConstants {
  96. public void run() {
  97. recvMessage();
  98. }
  99. public void recvMessage() {
  100. String name = Thread.currentThread().getName();
  101. try {
  102. Thread.sleep(1000);
  103. MQQueueManager qMgr = null;
  104. System.out.println("进入线程" + name);
  105. System.out.println(name + " Connecting to queue manager: "
  106. + qManager);
  107. qMgr = new MQQueueManager(qManager);
  108. // 设置将要连接的队列属性
  109. // Note. The MQC interface defines all the constants used by the
  110. // WebSphere MQ Java programming interface
  111. // (except for completion code constants and error code constants).
  112. // MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the
  113. // queue-defined default.
  114. // MQOO_OUTPUT:Open the queue to put messages.
  115. int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT
  116. | MQC.MQOO_INQUIRE;
  117. // Now get the message back again. First define a WebSphere MQ
  118. // message to receive the data
  119. MQMessage rcvMessage = new MQMessage();
  120. // Specify default get message options
  121. MQGetMessageOptions gmo = new MQGetMessageOptions();
  122. gmo.options = gmo.options + MQC.MQGMO_SYNCPOINT;// Get messages
  123. // under sync point
  124. // control（在同步点控制下获取消息）
  125. gmo.options = gmo.options + MQC.MQGMO_WAIT; // Wait if no messages
  126. // on the
  127. // Queue（如果在队列上没有消息则等待）
  128. gmo.options = gmo.options + MQC.MQGMO_FAIL_IF_QUIESCING;// Fail if
  129. // Qeue
  130. // Manager
  131. // Quiescing（如果队列管理器停顿则失败）
  132. gmo.waitInterval = 1000; // Sets the time limit for the
  133. // wait.（设置等待的毫秒时间限制）
  134. System.out.println(name + " Accessing queue: " + qName);
  135. MQQueue queue = qMgr.accessQueue(qName, openOptions);
  136. int depth = 0;
  137. // Get the message off the queue.
  138. System.out.println("... " + name + " getting the message back again");
  139. for (;;) {
  140. try {
  141. queue.get(rcvMessage, gmo);
  142. System.out.println(" ID: "
  143. + (new String(rcvMessage.messageId)).trim()
  144. + " Num: " + rcvMessage.messageSequenceNumber
  145. + " Type: " + rcvMessage.messageType + " Flag: "
  146. + rcvMessage.messageFlags);
  147. // And display the message text...
  148. String msgText = rcvMessage.readUTF();
  149. System.out.println("The message is: " + msgText);
  150. rcvMessage.clearMessage();
  151. // Break if no MSG left in queue
  152. depth = queue.getCurrentDepth();
  153. if (depth == 0)
  154. break;
  155. } catch (MQException mqe) {
  156. mqe.printStackTrace();
  157. break;
  158. // null;
  159. } catch (IOException e) {
  160. e.printStackTrace();
  161. }
  162. }
  163. // Close the queue
  164. System.out.println(name + " Closing the queue");
  165. queue.close();
  166. // Disconnect from the QueueManager
  167. System.out.println(name + " Disconnecting from the Queue Manager");
  168. qMgr.disconnect();
  169. System.out.println(name + " Done!");
  170. System.out.println("==========");
  171. System.out.println("");
  172. } catch (MQException ex) {
  173. System.out
  174. .println("A WebSphere MQ Error occured : Completion Code "
  175. + ex.completionCode + " Reason Code "
  176. + ex.reasonCode + ex.getMessage());
  177. } catch (InterruptedException e1) {
  178. e1.printStackTrace();
  179. }
  180. }
  181. }
  182. public class MQTest {
  183. public static void main(String args[]) {
  184. /*
  185. * MQTest first = new MQTest(); first.sendMessage();
  186. * first.recvMessage();
  187. */
  188. Sender sender = new Sender();
  189. Thread senderThread = new Thread(sender);
  190. senderThread.start();
  191. senderThread.setName("Sender");
  192. Thread senderThread2 = new Thread(sender);
  193. senderThread2.start();
  194. senderThread2.setName("Sender2");
  195. Receiver recv = new Receiver();
  196. Thread recvThread = new Thread(recv);
  197. recvThread.start();
  198. recvThread.setName("Receiver");
  199. // Receiver recv = new Receiver();
  200. // new Thread(recv).start();
  201. }
  202. }

运行结果如下：

进入线程Sender2
进入线程Sender
Sender2 Connecting to queue manager: XIR_QM_1502
Sender Connecting to queue manager: XIR_QM_1502
Sender2 Accessing queue: ESBREQ
Sender2 Sending a message...
Sender Accessing queue: ESBREQ
Sender Sending a message...
Sender2 putting the message... 1
Sender putting the message... 1
Sender2 putting the message... 2
Sender putting the message... 2
Sender2 putting the message... 3
Sender putting the message... 3
Sender2 putting the message... 4
Sender putting the message... 4
Sender2 Done!
==========

Sender Done!
==========

进入线程Receiver
Receiver Connecting to queue manager: XIR_QM_1502
Receiver Accessing queue: ESBREQ
... Receiver getting the message back again
 ID: MSGID Num: 1 Type: 1 Flag: 0
The message is: Salemetsizbe Yerasel 1
 ID: MSGID Num: 1 Type: 1 Flag: 0
The message is: Salemetsizbe Yerasel 1
 ID: MSGID Num: 1 Type: 1 Flag: 0
The message is: Salemetsizbe Yerasel 2
 ID: MSGID Num: 1 Type: 1 Flag: 0
The message is: Salemetsizbe Yerasel 2
 ID: MSGID Num: 1 Type: 1 Flag: 0
The message is: Salemetsizbe Yerasel 3
 ID: MSGID Num: 1 Type: 1 Flag: 0
The message is: Salemetsizbe Yerasel 3
 ID: MSGID Num: 1 Type: 1 Flag: 0
The message is: Salemetsizbe Yerasel 4
 ID: MSGID Num: 1 Type: 1 Flag: 0
The message is: Salemetsizbe Yerasel 4
Receiver Closing the queue
Receiver Disconnecting from the Queue Manager
Receiver Done!
==========

相关文章