MQI往队列写消息与取消息.docx

往队里写消息package com.iss;import java.io.BufferedWriter;import java.io.File;import java.io.FileInputStream;import java.io.FileWriter;import java.io.IOException;import java.sql.Connection;import java.sql.DriverManager;import java.sql.Statement;import java.text.SimpleDateFormat;import java.util.Date;import java.util.HashMap;import java.util.Iterator;import java.util.List;import java.util.Map;import java.util.Properties;import org.dom4j.DocumentHelper;import org.dom4j.Element;import com.ibm.mq.MQC;import com.ibm.mq.MQEnvironment;import com.ibm.mq.MQException;import com.ibm.mq.MQGetMessageOptions;import com.ibm.mq.MQMessage;import com.ibm.mq.MQQueue;import com.ibm.mq.MQQueueManager;public class MessageByMQ /定义队列管理器和队列的名称 private static String qmName; private static String qName; private static MQQueueManager qMgr; private static MQQueue queue; private static String getMessageQueueFlag;/队列： Y 表示常连接 N 表示非常连接用完就关闭 private static String getMessageQueueManagerFlag;/队列管理器： Y 表示常连接 N 表示非常连接用完就关闭 static try /设置环境: /MQEnvironment中包含控制MQQueueManager对象中的环境的构成的静态变量，MQEnvironment的值的设定会在MQQueueManager的构造函数加载的时候起作用， /load properties FileInputStream fis = new FileInputStream(new File(perties); Properties props = new Properties(); props.load(fis); fis.close(); /因此必须在建立MQQueueManager对象之前设定MQEnvironment中的值. getMessageQueueFlag=props.getProperty(getMessage.queue.queueAlwaysConnected); getMessageQueueManagerFlag=props.getProperty(getMessage.queue.queueManagerAlwaysConnected); MQEnvironment.hostname=props.getProperty(getMessage.queue.host);/MQ服务器的IP地址 MQEnvironment.channel=props.getProperty(getMessage.queue.channel);/服务器连接的通道 MQEnvironment.CCSID=Integer.valueOf(props.getProperty(getMessage.queue.ccsid);/服务器MQ服务使用的编码1381代表GBK、1208代表UTF(Coded Character Set Identifier:CCSID) MQEnvironment.port=Integer.valueOf(props.getProperty(getMessage.queue.port);/MQ端口 qmName = props.getProperty(getMessage.queue.qmname);/MQ的队列管理器名称 qName = props.getProperty(getMessage.queue.qname);/MQ远程队列的名称 /定义并初始化队列管理器对象并连接 /MQQueueManager可以被多线程共享，但是从MQ获取信息的时候是同步的，任何时候只有一个线程可以和MQ通信。 qMgr = new MQQueueManager(qmName); catch (MQException e) / TODO Auto-generated catch block System.out.println(初使化MQ出错); e.printStackTrace(); catch(Exception ex) System.out.println(初使化MQ出错); ex.printStackTrace(); private void getQueueInfo() throws Exception int count = 0; int count_error = 0; Connection con = null; try/ String url = jdbc:oracle:thin:37:1521:ORCL;/ 数据库连接字符串/ String username = cthi;/ 数据库用户名/ String password = cthi;/ 数据库密码/ / Class.forName(oracle.jdbc.driver.OracleDriver);/ con = DriverManager.getConnection(url, username, password); int openOptions = MQC.MQOO_BROWSE | MQC.MQOO_OUTPUT | MQC.MQOO_INQUIRE; MQGetMessageOptions gmo = new MQGetMessageOptions(); gmo.options = MQC.MQGMO_BROWSE_NEXT; / int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT | MQC.MQOO_INQUIRE;/ MQGetMessageOptions gmo = new MQGetMessageOptions(); / gmo.options = gmo.options + MQC.MQGMO_WAIT; / Wait if no messages on the Queue（如果在队列上没有消息则等待） / gmo.options = gmo.options + MQC.MQGMO_FAIL_IF_QUIESCING;/ Fail if Qeue Manager Quiescing（如果队列管理器停顿则失败） / gmo.waitInterval = 1 ; / Sets the time limit for the wait.（设置等待的毫秒时间限制） if(qMgr=null | !qMgr.isConnected() qMgr = new MQQueueManager(qmName); if(queue=null | !queue.isOpen() queue = qMgr.accessQueue(qName, openOptions); try String strMessage = ; File fs = new File(c:1abc_hr_error10707.txt); fs.createNewFile(); File f = new File(c:1abc_hr10707.txt); f.createNewFile(); FileWriter fw=null; BufferedWriter bw=null; SimpleDateFormat sdf = new SimpleDateFormat(yyyy-MM-dd HH:mm:ss); String sql = ;/Statement stmt = con.createStatement(); while(true) MQMessage retrieve = new MQMessage(); queue.get(retrieve, gmo); byte iii = new byteretrieve.getMessageLength(); retrieve.readFully(iii); / 取出队列的消息 strMessage = new String(iii,UTF-8);/ if(strMessage.trim().length() != 0)/ try/Element root = DocumentHelper.parseText(strMessage).getRootElement();/Element headElement = root.element(format);/List nodeLists = headElement.elements();/Map elementMap = new HashMap();/elementMap.put(classify, root.attributeValue(classify);/for(Iterator it = nodeLists.iterator();it.hasNext();)/Element element = (Element) it.next();/elementMap.put(element.getName(), element.getText();/sql =insert into MQCHECK (status,user_code,msg_id,insert_date) values(+String.valueOf(elementMap.get(classify)+,+String.valueOf(elementMap.get(owner)+,+String.valueOf(elementMap.get(id)+,sysdate);/stmt.executeUpdate(sql);/ catch(Exception ex)/ / /如果错误的消息格式放下边的文件当中/ count_error+;/ count-;/ FileWriter fws=null; / BufferedWriter bws=null; / fws = new FileWriter(fs, true);/ bws = new BufferedWriter(fws); / bws.write(strMessage); / bws.newLine(); / bws.flush(); / bws.close();/ ex.printStackTrace();/ / fw = new FileWriter(f, true); bw = new BufferedWriter(fw); bw.write(strMessage); bw.newLine(); bw.flush(); bw.close(); count+; catch(MQException ex) /当以MQC.MQOO_BROWSE(只读方式)读取消息时如果队列中消息已经被全部取走 mq 将返回 2033的code编码 可以捕捉此code 不做处理 catch(Exception ex) System.out.println(ServiceManageAction.getQueueInfo error); ex.printStackTrace(); throw ex; finally System.out.println(成功处理了 + count + 条); System.out.println(失败处理了 + count_error + 条); try queue.close(); catch (MQException e) e.printStackTrace(); try qMgr.disconnect(); catch (MQException e) e.printStackTrace(); / con.close(); /* * 从队列中去获取消息，如果队列中没有消息，就会发生异常，不过没有关系，有TRY.CATCH，如果是第三方程序调用方法，如果无返回则说明无消息 * 第三方可以将该方法放于一个无限循环的while(true).之中，不需要设置等待，因为在该方法内部在没有消息的时候会自动等待。 * return */ public static String getMessage() try /设置将要连接的队列属性 / Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface /(except for completion code constants and error code constants). /MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default. /MQOO_OUTPUT:Open the queue to put messages. int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT | MQC.MQOO_INQUIRE; /设置取出消息的属性（默认属性） /Set the put message options.（设置放置消息选项） MQGetMessageOptions gmo = new MQGetMessageOptions(); /gmo.options = gmo.options + MQC.MQGMO_SYNCPOINT;/Get messages under sync point control（在同步点控制下获取消息） gmo.options = gmo.options + MQC.MQGMO_WAIT; / Wait if no messages on the Queue（如果在队列上没有消息则等待） gmo.options = gmo.options + MQC.MQGMO_FAIL_IF_QUIESCING;/ Fail if Qeue Manager Quiescing（如果队列管理器停顿则失败） gmo.waitInterval = 1 ; / Sets the time limit for the wait.（设置等待的毫秒时间限制） /*关闭了就重新打开*/ if(qMgr=null | !qMgr.isConnected() qMgr = new MQQueueManager(qmName); /*要求为长连接所以保持一直打开状态*/ if(queue=null | !queue.isOpen() System.out.println(队列关闭了重新打开); queue = qMgr.accessQueue(qName, openOptions); / 从队列中取出消息 /先判断队列中是否有消息如果没有返回空字符串 if(getMessageCountByQ() = 0) return ;/队列中没有数据了返回空串 /一次读取队列中所有消息/ while(getMessageCountByQ() != 0)/ / MQMessage retrieve = new MQMessage();/ queue.get(retrieve, gmo);/ byte iii = new byteretrieve.getMessageLength();/ retrieve.readFully(iii); / 取出队列的消息/ messageList.add(new String(iii);/ /每调用一次读取一次队列中的消息 MQMessage retrieve = new MQMessage(); queue.get(retrieve, gmo);/ System.out.println(retrieve.correlationId+rrrrrrrrrrrrrr); byte iii = new byteretrieve.getMessageLength(); retrieve.readFully(iii); / 取出队列的消息 return new String(iii); catch (MQException ex) ex.printStackTrace(); System.out.println(Invoke get message. A WebSphere MQ error occurred : Completion code + pletionCode + Reason code + ex.reasonCode); catch (IOException ex) ex.printStackTrace(); System.out.println(Invoke get message. An error occurred whilst writing to the message buffer: + ex); catch(Exception ex) System.out.println(Invoke get message. An error); ex.printStackTrace(); finally if(getMessageQueueFlag.equalsIgnoreCase(N) try queue.close();/关闭队列 catch (MQException e) System.out.println(调用getMessage 后关闭队列失败！);e.printStackTrace(); if(getMessageQueueManagerFlag.equalsIgnoreCase(N) try qMgr.disconnect(); catch (MQException e) System.out.println(调用getMessage 后关闭队列管理器失败！); e.printStackTrace(); return ; /* * return int 返回消息条数 */ private static int getMessageCountByQ() int messagecount = 0; try messagecount = queue.getCurrentDepth(); catch (MQException e) System.out.println(queue.getCurrentDepth error!); e.printStackTrace(); return messagecount; /* * 关闭队列 * private static MQQueue queue; */ public static int closeQueueConnection() try queue.close(); catch (MQException e) System.out.println(调用 MessageByMQ.closeQueueConnection() 失败); e.printStackTrace(); return 2667;/关闭队列异常 catch(Exception e) System.out.println(调用 MessageByMQ.closeQueueConnection() 失败); e.printStackTrace(); return 2667;/关闭队列异常 return 1;/关闭成功 /* * 关闭队列管理器 * private static MQQueueManager qMgr; */ public static int closeQueueManagerConnection() try qMgr.disconnect(); catch (MQException e) System.out.println(调用 MessageByMQ.closeQueueManagerConnection() 失败); e.printStackTrace(); return 2671;/关闭队列管理器异常 catch(Exception e) System.out.println(调用 MessageByMQ.closeQueueManagerConnection() 失败); e.printStackTrace(); return 2671;/关闭队列管理器异常 return 1;/关闭成功 public static void main(String args) throws Exception /*下面两个方法可同时使用，也可以单独使用*/ System.out.println(MessageByMQ.closeQueueConnection();/ System.out.println(MessageSendMQ.closeQueueConnection();/ System.out.println(MessageProxy.sendMessage(aadsfasdfaf);/ System.out.println(MessageSendMQ.closeQueueConnection(); MessageByMQ messageByMQ = new MessageByMQ(); messageByMQ.getQueueInfo(); / String ss = 1234,5678;/ Element root = DocumentHelper.parseText(ss).getRootElement();/ System.out.println(root.getStringValue(); 取消息package com.iss;import java.io.File;import java.io.FileInputStream;import java.io.IOException;import java.util.List;import java.util.Properties;import com.ibm.mq.MQC;import com.ibm.mq.MQEnvironment;import com.ibm.mq.MQException;import com.ibm.mq.MQMessage;import com.ibm.mq.MQPutMessageOptions;import com.ibm.mq.MQQueue;import com.ibm.mq.MQQueueManager;public class MessageSendMQ /定义队列管理器和队列的名称 private static String qmName; private static String qName; private static MQQueueManager qMgr; private static MQQueue queue; private static String sendMessageQueueFlag;/队列：Y 表示常连接 N 表示非常连接用完就关闭 private static String sendMessageQueueManagerFlag;/队列管理器： Y 表示常连接 N 表示非常连接用完就关闭 static try /设置环境: /MQEnvironment中包含控制MQQueueManager对象中的环境的构成的静态变量，MQEnvironment的值的设定会在MQQueueManager的构造函数加载的时候起作用， /load properties FileInputStream fis = new FileInputStream(new File(perties); Properties props = new Properties(); props.load(fis); fis.close(); /因此必须在建立MQQueueManager对象之前设定MQEnvironment中的值. sendMessageQueueFlag=props.getProperty(sendMessage.queue.queueAlwaysConnected); sendMessageQueueManagerFlag=props.getProperty(sendMessage.queue.queueManagerAlwaysConnected); MQEnvironment.hostname=props.getProperty(sendMessage.queue.host);/MQ服务器的IP地址 MQEnvironment.channel=props.getProperty(sendMessage.queue.channel);/服务器连接的通道 MQEnvironment.CCSID=Integer.valueOf(props.getProperty(sendMessage.queue.ccsid);/服务器MQ服务使用的编码1381代表GBK、1208代表UTF(Coded Character Set Identifier:CCSID) MQEnvironment.port=Integer.valueOf(props.getProperty(sendMessage.queue.port);/MQ端口 qmName = props.getProperty(sendMessage.queue.qmname);/MQ的队列管理器名称 qName = props.getProperty(sendMessage.queue.qname);/MQ远程队列的名称 /定义并初始化队列管理器对象并连接 /MQQueueManager可以被多线程共享，但是从MQ获取信息的时候是同步的，任何时候只有一个线程可以和MQ通信。 qMgr = new MQQueueManager(qmName); catch (MQException e) / TODO Auto-generated catch block System.out.println(初使化MQ出错); e.printStackTrace(); catch(Exception ex) System.out.println(初使化MQ出错); ex.printStackTrace(); /* * 往MQ发送消息 * param message * return */ public static int sendMessage(String message) int result=0;/代表发送失败初始化为0 try /设置将要连接的队列属性 / Note. The MQC interface defines all the constants used by the WebSphere MQ Java programming interface /(except for completion code constants and error code constants). /MQOO_INPUT_AS_Q_DEF:Open the queue to get messages using the queue-defined default. /MQOO_OUTPUT:Open the queue to put messages. /*目标为远程队列，所有这里不可以用MQOO_INPUT_AS_Q_DEF属性*/ /int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT; /*以下选项可适合远程队列与本地队列*/ int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING | MQC.MQOO_INQUIRE; /连接队列 /MQQueue provides inquire, set, put and get operations for WebSphere MQ queues. /The inquire and set capabilities are inherited from MQManagedObject. /*关闭了就重新打开*/ if(qMgr=null | !qMgr.isConnected() qMgr = new MQQueueManager(qmName); /*要求为长连接所以保持一直打开状态*/ if(queue=null | !queue.isOpen() queue = qMgr.accessQueue(qName, openOptions); /定义一个简单的消息 MQMessage putMessage = new MQMessage(); /将数据放入消息缓冲区 putMessage.characterSet=1381; putMessage.write(message.getBytes(); /设置写入消息的属性（默认属性） MQPutMessageOptions pmo = new MQPutMessageOptions()