java结合WebSphere MQ实现接收队列文件功能

WebSphereMQ，也称MQSeries，以一致的、可靠的和易于管理的方式来连接应用程序，并为跨部门、企业范围的集成提供了可靠的基础。通过为重要的消息和事务提供可靠的、一次且仅一次的传递，MQ可以处理复杂的通信协议，并动态地将消息传递工作负载分配给可用的资源。

首先我们先来简单介绍下websphere mq以及安装使用简介

websphere mq : 用于传输信息 具有跨平台的功能。

1 安装websphere mq 并启动

2 websphere mq 建立 queue Manager (如：MQSI_SAMPLE_QM)

3 建立queue 类型选择 Local类型 的 (如lq )

4建立channels 类型选择Server Connection (如BridgeChannel)

接下来，我们来看实例代码：

1. MQFileReceiver.java 
2.  
3. package com.mq.dpca.file; 
4.  
5. import java.io.File; 
6. import java.io.FileOutputStream; 
7.  
8. import com.ibm.mq.MQEnvironment; 
9. import com.ibm.mq.MQException; 
10. import com.ibm.mq.MQGetMessageOptions; 
11. import com.ibm.mq.MQMessage; 
12. import com.ibm.mq.MQQueue; 
13. import com.ibm.mq.MQQueueManager; 
14. import com.ibm.mq.constants.MQConstants; 
15. import com.mq.dpca.msg.MQConfig; 
16. import com.mq.dpca.util.ReadCmdLine; 
17. import com.mq.dpca.util.RenameUtil; 
18.  
19. /** 
20. *  
21. * MQ分组接收文件功能 
22. * 主动轮询 
23. */ 
24. public class MQFileReceiver { 
25. private MQQueueManager qmgr; // 连接到队列管理器 
26.  
27. private MQQueue inQueue; // 传输队列 
28.  
29. private String queueName = ""; // 队列名称 
30.  
31. private String host = ""; // 
32.  
33. private int port = 1414; // 侦听器的端口号 
34.  
35. private String channel = ""; // 通道名称 
36.  
37. private String qmgrName = ""; // 队列管理器 
38.  
39. private MQMessage inMsg; // 创建消息缓冲 
40.  
41. private MQGetMessageOptions gmo; // 设置获取消息选项 
42.  
43. private static String fileName = null; // 接收队列上的消息并存入文件 
44.  
45. private int ccsid = 0; 
46.  
47. private static String file_dir = null; 
48.  
49. /** 
50. * 程序的入口 
51. *  
52. * @param args 
53. */ 
54. public static void main(String args[]) { 
55. MQFileReceiver mfs = new MQFileReceiver(); 
56. //初始化连接 
57. mfs.initproperty(); 
58. //接收文件 
59. mfs.runGoupReceiver(); 
60. //获取shell脚本名 
61. // String shellname = MQConfig.getValueByKey(fileName); 
62. // if(shellname!=null&&!"".equals(shellname)){ 
63. // //调用shell 
64. // ReadCmdLine.callShell(shellname); 
65. // }else{ 
66. // System.out.println("have no shell name,Only receive files."); 
67. // } 
68.  
69. } 
70.  
71. public void runGoupReceiver() { 
72. try { 
73. init(); 
74. getGroupMessages(); 
75. qmgr.commit(); 
76. System.out.println("/n Messages successfully Receive "); 
77. } catch (MQException mqe) { 
78. mqe.printStackTrace(); 
79. try { 
80. System.out.println("/n Backing out Transaction "); 
81. qmgr.backout(); 
82. System.exit(2); 
83. } catch (Exception e) { 
84. e.printStackTrace(); 
85. System.exit(2); 
86. } 
87. } catch (Exception e) { 
88. e.printStackTrace(); 
89. System.exit(2); 
90. } 
91. } 
92.  
93. /** 
94. * 初始化服务器连接信息 
95. *  
96. * @throws Exception 
97. */ 
98. private void init() throws Exception { 
99. /* 为客户机连接设置MQEnvironment属性 */ 
100. MQEnvironment.hostname = host; 
101. MQEnvironment.channel = channel; 
102. MQEnvironment.port = port; 
103.  
104. /* 连接到队列管理器 */ 
105. qmgr = new MQQueueManager(qmgrName); 
106.  
107. /* 设置队列打开选项以输 */ 
108. int opnOptn = MQConstants.MQOO_INPUT_AS_Q_DEF 
109. | MQConstants.MQOO_FAIL_IF_QUIESCING; 
110.  
111. /* 打开队列以输 */ 
112. inQueue = qmgr.accessQueue(queueName, opnOptn, null, null, null); 
113. } 
114.  
115. /** 
116. * 接受文件的主函数 
117. *  
118. * @throws Exception 
119. */ 
120. public void getGroupMessages() { 
121. /* 设置获取消息选项 */ 
122. gmo = new MQGetMessageOptions(); 
123. gmo.options = MQConstants.MQGMO_FAIL_IF_QUIESCING; 
124. gmo.options = gmo.options + MQConstants.MQGMO_SYNCPOINT; 
125. /* 等待消息 */ 
126. gmo.options = gmo.options + MQConstants.MQGMO_WAIT; 
127. /* 设置等待时间限制 */ 
128. gmo.waitInterval = 5000; 
129. /* 只获取消息 */ 
130. gmo.options = gmo.options + MQConstants.MQGMO_ALL_MSGS_AVAILABLE; 
131. /* 以辑顺序获取消息 */ 
132. gmo.options = gmo.options + MQConstants.MQGMO_LOGICAL_ORDER; 
133. gmo.matchOptions = MQConstants.MQMO_MATCH_GROUP_ID; 
134. /* 创建消息缓冲 */ 
135. inMsg = new MQMessage(); 
136. try { 
137. FileOutputStream fos = null; 
138. /* 处理组消息 */ 
139. while (true) { 
140. try { 
141. inQueue.get(inMsg, gmo); 
142. if (fos == null) { 
143. try { 
144. fileName = inMsg.getStringProperty("fileName"); 
145. String fileName_full = null; 
146. fileName_full = file_dir + RenameUtil.rename(fileName); 
147. fos = new FileOutputStream(new File(fileName_full)); 
148. int msgLength = inMsg.getMessageLength(); 
149. byte[] buffer = new byte[msgLength]; 
150. inMsg.readFully(buffer); 
151. fos.write(buffer, 0, msgLength); 
152. /* 查看是否是最后消息标识 */ 
153. char x = gmo.groupStatus; 
154. if (x == MQConstants.MQGS_LAST_MSG_IN_GROUP) { 
155. System.out.println("Last Msg in Group"); 
156. break; 
157. } 
158. inMsg.clearMessage(); 
159.  
160. } catch (Exception e) { 
161. System.out 
162. .println("Receiver the message without property,do nothing!"); 
163. inMsg.clearMessage(); 
164. } 
165. } else { 
166. int msgLength = inMsg.getMessageLength(); 
167. byte[] buffer = new byte[msgLength]; 
168. inMsg.readFully(buffer); 
169. fos.write(buffer, 0, msgLength); 
170. /* 查看是否是最后消息标识 */ 
171. char x = gmo.groupStatus; 
172. if (x == MQConstants.MQGS_LAST_MSG_IN_GROUP) { 
173. System.out.println("Last Msg in Group"); 
174. break; 
175. } 
176. inMsg.clearMessage(); 
177. } 
178. } catch (Exception e) { 
179. char x = gmo.groupStatus; 
180. if (x == MQConstants.MQGS_LAST_MSG_IN_GROUP) { 
181. System.out.println("Last Msg in Group"); 
182. } 
183. break; 
184. } 
185. } 
186. if (fos != null) 
187. fos.close(); 
188. } catch (Exception e) { 
189. System.out.println(e.getMessage()); 
190. } 
191. } 
192.  
193. public void initproperty() { 
194. MQConfig config = new MQConfig().getInstance(); 
195. if (config.getMQ_MANAGER() != null) { 
196. qmgrName = config.getMQ_MANAGER(); 
197. queueName = config.getMQ_QUEUE_NAME(); 
198. channel = config.getMQ_CHANNEL(); 
199. host = config.getMQ_HOST_NAME(); 
200. port = Integer.valueOf(config.getMQ_PROT()); 
201. ccsid = Integer.valueOf(config.getMQ_CCSID()); 
202. file_dir = config.getFILE_DIR(); 
203. } 
204. } 
205. }