IBM MQ的java实现例子

第一次接触MQ(Message Query)消息队列，就是往队列里面写消息和从队列里面读取消息两种方式

 

package com.bulain.wasmq;   
  
import org.apache.log4j.Logger;   
  
import java.io.InputStream;   
import java.util.Properties;   
  
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 Main {   
    /**  
     * Logger for this class  
     */  
    private static final Logger logger = Logger.getLogger(Main.class);   
       
    private String strExtraSendXmlFileName = "jndi.properties";   
    private static Properties props;   
    static {   
        props = new Properties();   
        props.put("mqHostName","10.240.13.81");   
        props.put("mqPort","1414");   
        props.put("mqCCSID","932");        
        props.put("mqUserName","mqadmin");   
        props.put("mqPassword","Art515940");   
        props.put("mqQManager","QM_mphch085");   
        props.put("mqChannel","S_mphch085");   
        props.put("mqLocalOutQueue","clq_default_mphch085");   
        props.put("mqLocalInQueue","clq_default_mphch085");        
    }   
       
  
    public static void main(String[] args) {   
        Main test = new Main();   
        test.send();   
        test.recieve();   
    }   
  
    public void send() {   
        // MQ?送   
        try {   
            // 建立MQ客?端?用上下文?境   
            MQEnvironment.hostname = props.getProperty("mqHostName"); // 服?器ip地址   
            MQEnvironment.port = Integer.parseInt(props.getProperty("mqPort")); // 服?器MQ服?端口   
            MQEnvironment.CCSID = Integer.parseInt(props.getProperty("mqCCSID")); // 服?器MQ服?使用的??   
            MQEnvironment.channel = props.getProperty("mqChannel"); // 服?器?接通道名   
            MQEnvironment.userID = props.getProperty("mqUserName"); // MQ服?用?名   
            MQEnvironment.password = props.getProperty("mqPassword");   
  
            // ?接?列管理器   
            MQQueueManager qMgr = new MQQueueManager(props.getProperty("mqQManager"));   
            int openOptions = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING;   
            // 打?MQ?列   
            MQQueue q = qMgr.accessQueue(props.getProperty("mqLocalOutQueue"), openOptions);   
            //FileInputStream fins = new FileInputStream(new File(strExtraSendXmlFileName));   
            InputStream fins = ClassLoader.getSystemResourceAsStream(strExtraSendXmlFileName);   
            byte[] data = new byte[fins.available()];   
            fins.read(data);   
            fins.close();   
            MQMessage msg = new MQMessage();   
            msg.write(data);   
            // 放入消息   
            q.put(msg);   
            // ???列   
            q.close();   
            // ???列管理器   
            qMgr.disconnect();   
        } catch (MQException e) {   
            logger.error(e);   
            e.printStackTrace();   
        } catch (Exception e) {   
            logger.error(e);   
            e.printStackTrace();   
        }   
    }   
  
    public void recieve() {   
        // MQ接收   
        try {   
            // 建立MQ客?端?用上下文?境   
            MQEnvironment.hostname = props.getProperty("mqHostName"); // 服?器ip地址   
            MQEnvironment.port = Integer.parseInt(props.getProperty("mqPort")); // 服?器MQ服?端口   
            MQEnvironment.CCSID = Integer.parseInt(props.getProperty("mqCCSID")); // 服?器MQ服?使用的??   
            MQEnvironment.channel = props.getProperty("mqChannel"); // 服?器?接通道名   
            MQEnvironment.userID = props.getProperty("mqUserName"); // MQ服?用?名   
            MQEnvironment.password = props.getProperty("mqPassword");   
  
            // ?接?列管理器   
            MQQueueManager qMgr = new MQQueueManager(props.getProperty("mqQManager"));   
            int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_FAIL_IF_QUIESCING;   
            // 打?MQ?列   
            MQQueue q = qMgr.accessQueue(props.getProperty("mqLocalInQueue"), openOptions);   
  
            MQGetMessageOptions mgo = new MQGetMessageOptions();   
            mgo.options |= MQC.MQGMO_NO_WAIT;   
  
            MQMessage msg = new MQMessage();   
  
            if ((msg = fetchOneMsg(q)) != null) {   
                byte[] xmlData = new byte[msg.getDataLength()];   
                msg.readFully(xmlData);   
                logger.info(new String(xmlData));   
            }   
            // ???列   
            q.close();   
            // ???列管理器   
            qMgr.disconnect();   
        } catch (MQException e) {   
            logger.error(e);   
            e.printStackTrace();   
        } catch (Exception e) {   
            logger.error(e);   
            e.printStackTrace();   
        }   
    }   
  
    /**  
     * 从?列中取出一个消息  
     *   
     * @param q  
     *            ?列名称  
     * @return  
     * @throws Exception  
     */  
    private static MQMessage fetchOneMsg(MQQueue q) throws Exception {   
        MQGetMessageOptions mgo = new MQGetMessageOptions();   
        mgo.options |= MQC.MQGMO_NO_WAIT;   
        MQMessage msg = new MQMessage();   
        try {   
            // ?取消息   
            q.get(msg, mgo);   
        } catch (MQException e) {   
            return null;   
        }   
        return msg;   
    }   
}