1 import java.io.IOException;
2 import java.util.HashMap;
3 import java.util.Map;
4
5 import com.ibm.mq.MQC;
6 import com.ibm.mq.MQEnvironment;
7 import com.ibm.mq.MQException;
8 import com.ibm.mq.MQGetMessageOptions;
9 import com.ibm.mq.MQMessage;
10 import com.ibm.mq.MQPutMessageOptions;
11 import com.ibm.mq.MQQueue;
12 import com.ibm.mq.MQQueueManager;
13
14 public class CLIENT_MQ{
15 //定义队列管理器和队列的名称
16 private static final String qmName = "MQ_SERVICE";//MQ的队列管理器名称 ;
17 //private static final String qName = "MIDDLE_SEND_QUEUE"; //MQ远程队列的名称
18 private static MQQueueManager qMgr;//队列管理器
19 public static void init(){
20 //设置环境:
21 //MQEnvironment中包含控制MQQueueManager对象中的环境的构成的静态变量，MQEnvironment的值的设定会在MQQueueManager的构造函数加载的时候起作用，
22 //因此必须在建立MQQueueManager对象之前设定MQEnvironment中的值.
23 MQEnvironment.hostname="10.172.12.156"; //MQ服务器的IP地址
24 MQEnvironment.channel="SERVICE_JAVA"; //通道类型:服务器连接
25 MQEnvironment.CCSID=1381;//437 //服务器MQ服务使用的编码1381代表GBK、1208代表UTF(Coded Character Set Identifier:CCSID)
26 MQEnvironment.port=1456; //MQ端口
27 try {
28 //定义并初始化队列管理器对象并连接
29 //MQQueueManager可以被多线程共享，但是从MQ获取信息的时候是同步的，任何时候只有一个线程可以和MQ通信。
30 qMgr = new MQQueueManager(qmName);
31 } catch (MQException e) {
32 // TODO Auto-generated catch block
33 System.out.println("初使化MQ出错");
34 e.printStackTrace();
35 }
36 }
37 /**
38 * 往MQ发送消息
39 * @param message
40 * @return
41 */
42 public static Map<String,Object> sendMessage(Object message,String qName){
43 Map<String,Object> map=new HashMap<String,Object>();
44 try{
45 //设置将要连接的队列属性
46 // Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface
47 //(except for completion code constants and error code constants).
48 //MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default.
49 //MQOO_OUTPUT:Open the queue to put messages.
50 /*目标为远程队列，所有这里不可以用MQOO_INPUT_AS_Q_DEF属性*/
51 //int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT;
52 /*以下选项可适合远程队列与本地队列*/
53 //int openOptions = MQC.MQOO_INQUIRE | MQC.MQOO_OUTPUT; //发送时使用
54 //int qOptioin = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT; 接收时使用
55 int openOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING;
56 //连接队列
57 //MQQueue provides inquire, set, put and get operations for WebSphere MQ queues.
58 //The inquire and set capabilities are inherited from MQManagedObject.
59 /*关闭了就重新打开*/
60 if(qMgr==null || !qMgr.isConnected()){
61 qMgr = new MQQueueManager(qmName);
62 }
63 MQQueue queue = qMgr.accessQueue(qName, openOptions);
64 //定义一个简单的消息
65 MQMessage putMessage = new MQMessage();
66 map.put("messageId",putMessage);
67 //String uuid=java.util.UUID.randomUUID().toString();
68 //将数据放入消息缓冲区
69 putMessage.writeObject(message);
70 //设置写入消息的属性（默认属性）
71 MQPutMessageOptions pmo = new MQPutMessageOptions();
72
73 //将消息写入队列
74 queue.put(putMessage,pmo);
75 map.put("message",message.toString());
76 queue.close();
77 }catch (MQException ex) {
78 System.out.println("A WebSphere MQ error occurred : Completion code " + ex.completionCode + " Reason code " + ex.reasonCode);
79 ex.printStackTrace();
80 }catch (IOException ex) {
81 System.out.println("An error occurred whilst writing to the message buffer: " + ex);
82 }catch(Exception ex){
83 ex.printStackTrace();
84 }finally{
85 try {
86 qMgr.disconnect();
87 } catch (MQException e) {
88 e.printStackTrace();
89 }
90 }
91 return map;
92 }
93
94
95
96
97
98 /**
99 * 处理完消息回放到MQ队列
100 * @param message
101 * @return
102 */
103 public static Map<String,Object> sendReplyMessage(Object message,String qName,MQMessage mqMessage){
104 Map<String,Object> map=new HashMap<String,Object>();
105 try{
106 //设置将要连接的队列属性
107 // Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface
108 //(except for completion code constants and error code constants).
109 //MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default.
110 //MQOO_OUTPUT:Open the queue to put messages.
111 /*目标为远程队列，所有这里不可以用MQOO_INPUT_AS_Q_DEF属性*/
112 //int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT;
113 /*以下选项可适合远程队列与本地队列*/
114 //int openOptions = MQC.MQOO_INQUIRE | MQC.MQOO_OUTPUT; //发送时使用
115 //int qOptioin = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT; 接收时使用
116 int openOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING;
117 //连接队列
118 //MQQueue provides inquire, set, put and get operations for WebSphere MQ queues.
119 //The inquire and set capabilities are inherited from MQManagedObject.
120 /*关闭了就重新打开*/
121 if(qMgr==null || !qMgr.isConnected()){
122 qMgr = new MQQueueManager(qmName);
123 }
124 MQQueue queue = qMgr.accessQueue(qName, openOptions);
125 //定义一个简单的消息
126 MQMessage putMessage = new MQMessage();
127 putMessage.messageId=mqMessage.messageId;
128 map.put("messageId",putMessage);
129 //String uuid=java.util.UUID.randomUUID().toString();
130 //将数据放入消息缓冲区
131 putMessage.writeObject(message);
132 //设置写入消息的属性（默认属性）
133 MQPutMessageOptions pmo = new MQPutMessageOptions();
134
135 //将消息写入队列
136 queue.put(putMessage,pmo);
137 map.put("message",message.toString());
138 queue.close();
139 }catch (MQException ex) {
140 System.out.println("A WebSphere MQ error occurred : Completion code " + ex.completionCode + " Reason code " + ex.reasonCode);
141 ex.printStackTrace();
142 }catch (IOException ex) {
143 System.out.println("An error occurred whilst writing to the message buffer: " + ex);
144 }catch(Exception ex){
145 ex.printStackTrace();
146 }finally{
147 try {
148 qMgr.disconnect();
149 } catch (MQException e) {
150 e.printStackTrace();
151 }
152 }
153 return map;
154 }
155
156
157
158
159 /**
160 * 从队列中去获取消息，如果队列中没有消息，就会发生异常，不过没有关系，有TRY...CATCH，如果是第三方程序调用方法，如果无返回则说明无消息
161 * 第三方可以将该方法放于一个无限循环的while(true){...}之中，不需要设置等待，因为在该方法内部在没有消息的时候会自动等待。
162 * @return
163 */
164 public static String getMessage(String qName,MQMessage mqMessage){
165 String message="";
166 try{
167 //设置将要连接的队列属性
168 // Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface
169 //(except for completion code constants and error code constants).
170 //MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default.
171 //MQOO_OUTPUT:Open the queue to put messages.
172 //int qOptioin = MQC.MQOO_INQUIRE | MQC.MQOO_OUTPUT; 发送时使用
173 //int qOptioin = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT; 接收时使用
174
175 int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT;
176 MQMessage retrieve = new MQMessage();
177 //设置取出消息的属性（默认属性）
178 //Set the put message options.（设置放置消息选项）
179 MQGetMessageOptions gmo = new MQGetMessageOptions();
180
181 gmo.options = gmo.options + MQC.MQGMO_SYNCPOINT;//Get messages under sync point control（在同步点控制下获取消息）
182 gmo.options = gmo.options + MQC.MQGMO_WAIT; // Wait if no messages on the Queue（如果在队列上没有消息则等待）
183 gmo.options = gmo.options + MQC.MQGMO_FAIL_IF_QUIESCING;// Fail if Qeue Manager Quiescing（如果队列管理器停顿则失败）
184 gmo.waitInterval = 3000 ; // Sets the time limit for the wait.（设置等待的毫秒时间限制）
185 /*关闭了就重新打开*/
186 if(qMgr==null || !qMgr.isConnected()){
187 qMgr = new MQQueueManager(qmName);
188 }
189 MQQueue queue = qMgr.accessQueue(qName, openOptions);
190
191 MQMessage retrievedMessage = new MQMessage();
192 //从队列中取出对应messageId的消息
193 retrieve.messageId = mqMessage.messageId;
194 // 从队列中取出消息
195 queue.get(retrieve, gmo);
196
197
198 Object obj = retrieve.readObject();
199 message=obj.toString();//解决中文乱码问题
200 /*
201
202 //int size = rcvMessage.getMessageLength();
203 //byte[] p = new byte[size];
204 //rcvMessage.readFully(p);
205
206 int len=retrieve.getDataLength();
207 byte[] str = new byte[len];
208 retrieve.readFully(str,0,len);
209 message = new String(str);//readUTF();
210 */
211
212 queue.close();
213 }catch (MQException ex) {
214 int reason=ex.reasonCode;
215 if(reason==2033)//no messages
216 {
217 message="nomessage";
218 }else{
219 System.out.println("A WebSphere MQ error occurred : Completion code " + ex.completionCode + " Reason code " + ex.reasonCode);
220 }
221 }catch (IOException ex) {
222 System.out.println("An error occurred whilst writing to the message buffer: " + ex);
223 }catch(Exception ex){
224 ex.printStackTrace();
225 }finally{
226 try {
227 qMgr.disconnect();
228 } catch (MQException e) {
229 e.printStackTrace();
230 }
231 }
232 return message;
233 }
234
235
236
237 public static void main(String args[]) {
238 init();
239 Map<String,Object> map = new HashMap<String,Object>();
240 map=sendMessage("{name: test get message id 123}","SERVICE_TRANSFER_QUEUE");
241 MQMessage mqMessage = (MQMessage)map.get("messageId");
242 outSys("传输消息:",mqMessage.messageId.toString());
243
244 outSys("接收传输队列:",getMessage("SERVICE_TRANSFER_QUEUE",mqMessage));
245 Map<String,Object> reply_map = new HashMap<String,Object>();
246 reply_map=sendReplyMessage("{name: local queue 008}","SERVICE_RECEIVE_QUEUE",mqMessage);
247 outSys("放入正常队列:",reply_map.get("message").toString());
248
249 outSys("接收正常队列:",getMessage("SERVICE_RECEIVE_QUEUE",mqMessage));
250
251
252 }
253
254
255
256 public static void outSys(String display,String val){
257 System.out.println(display+val);
258 }
259
260 }
时间： 2024-10-06 03:05:17