java往MQ里面发消息和取消息
1package test; 2 import java.io.IOException; 3 4import com.ibm.mq.MQC; 5import com.ibm.mq.MQEnvironment; 6import com.ibm.mq.MQException; 7import com.ibm.mq.MQGetMessageOptions; 8import com.ibm.mq.MQMessage; 9import com.ibm.mq.MQPutMessageOptions; 10import com.ibm.mq.MQQueue; 11import com.ibm.mq.MQQueueManager; 12import com.ibm.mq.constants.MQConstants; 13 14 public class test1{ 15 //定义队列管理器和队列的名称 16 private static String qmName; 17 private static String qName; 18 private static MQQueueManager qMgr; 19 static{ 20 //设置环境: 21 //MQEnvironment中包含控制MQQueueManager对象中的环境的构成的静态变量，MQEnvironment的值的设定会在MQQueueManager的构造函数加载的时候起作用， 22 //因此必须在建立MQQueueManager对象之前设定MQEnvironment中的值. 23 MQEnvironment.hostname="172.31.17.162"; //MQ服务器的IP地址 24 MQEnvironment.channel="SVRCONN_GW"; //服务器连接的通道 25 MQEnvironment.CCSID=1208; //服务器MQ服务使用的编码1381代表GBK、1208代表UTF(Coded Character Set Identifier:CCSID) 26 MQEnvironment.port=33333; //MQ端口 27 qmName = "QM_TEST"; //MQ的队列管理器名称 28 qName = "TEST"; //MQ远程队列的名称 29 try { 30 //定义并初始化队列管理器对象并连接 31 //MQQueueManager可以被多线程共享，但是从MQ获取信息的时候是同步的，任何时候只有一个线程可以和MQ通信。 32 qMgr = new MQQueueManager(qmName); 33 } catch (MQException e) { 34 // TODO Auto-generated catch block 35 System.out.println("初使化MQ出错"); 36 e.printStackTrace(); 37 } 38 } 39 /** 40 * 往MQ发送消息 41 * @param message 42 * @return 43 */ 44 public static int sendMessage(String message){ 45 int result=0; 46 try{ 47 //设置将要连接的队列属性 48 // Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface 49 //(except for completion code constants and error code constants). 50 //MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default. 51 //MQOO_OUTPUT:Open the queue to put messages. 52 /*目标为远程队列，所有这里不可以用MQOO_INPUT_AS_Q_DEF属性*/ 53 //int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT; 54 /*以下选项可适合远程队列与本地队列*/ 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 //将数据放入消息缓冲区 67 putMessage.writeUTF(message); 68 //设置写入消息的属性（默认属性） 69 MQPutMessageOptions pmo = new MQPutMessageOptions(); 70 //将消息写入队列 71 queue.put(putMessage,pmo); 72 queue.close(); 73 }catch (MQException ex) { 74 System.out.println("A WebSphere MQ error occurred : Completion code " 75 + ex.completionCode + " Reason code " + ex.reasonCode); 76 ex.printStackTrace(); 77 }catch (IOException ex) { 78 System.out.println("An error occurred whilst writing to the message buffer: " + ex); 79 }catch(Exception ex){ 80 ex.printStackTrace(); 81 }finally{ 82 try { 83 qMgr.disconnect(); 84 } catch (MQException e) { 85 e.printStackTrace(); 86 } 87 } 88 return result; 89 } 90 /** 91 * 从队列中去获取消息，如果队列中没有消息，就会发生异常，不过没有关系，有TRY...CATCH，如果是第三方程序调用方法，如果无返回则说明无消息 92 * 第三方可以将该方法放于一个无限循环的while(true){...}之中，不需要设置等待，因为在该方法内部在没有消息的时候会自动等待。 93 * @return 94 */ 95 public static String getMessage(){ 96 String message=null; 97 try{ 98 //设置将要连接的队列属性 99 // Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface 100 //(except for completion code constants and error code constants). 101 //MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default. 102 //MQOO_OUTPUT:Open the queue to put messages. 103 int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT; 104 MQMessage retrieve = new MQMessage(); 105 //设置取出消息的属性（默认属性） 106 //Set the put message options.（设置放置消息选项） 107 MQGetMessageOptions gmo = new MQGetMessageOptions(); 108 gmo.options = gmo.options + MQC.MQGMO_SYNCPOINT;//Get messages under sync point control（在同步点控制下获取消息） 109 gmo.options = gmo.options + MQC.MQGMO_WAIT; // Wait if no messages on the Queue（如果在队列上没有消息则等待） 110 gmo.options = gmo.options + MQC.MQGMO_FAIL_IF_QUIESCING;// Fail if Qeue Manager Quiescing（如果队列管理器停顿则失败） 111 gmo.waitInterval = 1000 ; // Sets the time limit for the wait.（设置等待的毫秒时间限制） 112 /*关闭了就重新打开*/ 113 if(qMgr==null || !qMgr.isConnected()){ 114 qMgr = new MQQueueManager(qmName); 115 } 116 MQQueue queue = qMgr.accessQueue(qName, openOptions); 117 // 从队列中取出消息 118 queue.get(retrieve, gmo); 119 byte[] buf = new byte[retrieve.getMessageLength()]; 120 retrieve.readFully(buf); 121 122 System.out.println("The message is: " + new String(buf)); 123 queue.close(); 124 }catch (MQException ex) { 125 System.out.println("A WebSphere MQ error occurred : Completion code " 126 + ex.completionCode + " Reason code " + ex.reasonCode); 127 }catch (IOException ex) { 128 System.out.println("An error occurred whilst writing to the message buffer: " + ex); 129 }catch(Exception ex){ 130 ex.printStackTrace(); 131 }finally{ 132 try { 133 qMgr.disconnect(); 134 } catch (MQException e) 135 { e.printStackTrace(); 136 } 137 } 138 return message; 139 } 140 public static void main(String args[]) throws InterruptedException { 141 /*下面两个方法可同时使用，也可以单独使用*/ 142 //sendMessage("this is a test"); 143 //System.out.println("jwh"); 144 for (int i = 0; i < 100; i++) { 145 getMessage(); 146 Thread.sleep(20000); 147 } 148 149 150 151 } 152 } 153 *