IBM MQ 使用一例

qiantin

MQ作为一种中间件还是不错的，可以由一个系统发送一些消息给MQ Server，然后另外一个系统来取得这些消息。比较方便的实现不同系统、不同语言间的通信。

MQ Server可以进行消息的排队、缓存等。

具体MQ Server的配置我就不介绍了，可以去IBM的网站下载一个试用版试试看。

下面是一个收发的简单的小例子，具体意思看里面的注释。


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.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;

/**//*
* 可以在MQ的资源管理器的某一个队列上放入测试消息、浏览消息等
* 可以放入多条消息，按先进先出的方式取得
*/
public class MQTest {
   
    private String qManager;// QueueManager名

    private MQQueueManager qMgr;

    private MQQueue qQueue;
   
    String HOST_NAME;
   
    int PORT=0;

    String Q_NAME;
   
    String CHANNEL;
   
    int CCSID;
   
    String Msg;
   
    public void init(){
            
        try {
            HOST_NAME="192.168.88.169";
            PORT=1414;
            qManager="QM_yfgaochun";
            Q_NAME="QM_NAME1";
            CHANNEL="S_yfgaochun";
            CCSID=1381; //表示是简体中文， CCSID的值在AIX上一般设为1383，如果要支持GBK则设为1386，在WIN上设为1381。
            
            Msg = "MQ测试发送Gao";
            //System.out.println(Msg);
            
            MQEnvironment.hostname = HOST_NAME; //安裝MQ所在的ip address
            MQEnvironment.port=PORT; // TCP/IP port
            
            MQEnvironment.channel=CHANNEL;
            MQEnvironment.CCSID=CCSID;
            
            qMgr = new MQQueueManager(qManager);
            
/**//*            try {
                XADataSource ds = null;
                Connection con = (Connection)qMgr.getJDBCConnection(ds);
            } catch (SQLException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            }*/
            
            //int qOptioin = MQC.MQOO_INQUIRE | MQC.MQOO_OUTPUT; 发送时使用
            //int qOptioin = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT; 接收时使用
            int qOptioin = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_INQUIRE | MQC.MQOO_OUTPUT;

            qQueue = qMgr.accessQueue(Q_NAME, qOptioin);
            
        } catch (MQException e) {
            System.out
                    .println("A WebSphere MQ error occurred : Completion code "
                            + e.completionCode + " Reason Code is "
                            + e.reasonCode);
        }
    }
   
    void finalizer() {
        try {
            qQueue.close();
            qMgr.disconnect();
        } catch (MQException e) {
            System.out
                    .println("A WebSphere MQ error occurred : Completion code "
                            + e.completionCode + " Reason Code is "
                            + e.reasonCode);
        }
    }

    /**//*
     * 取过一次，下次就没有了
     */
    public void GetMsg() {
        try {
            MQMessage retrievedMessage = new MQMessage();

            MQGetMessageOptions gmo = new MQGetMessageOptions();
            gmo.options += MQC.MQPMO_SYNCPOINT;

            qQueue.get(retrievedMessage, gmo);

            int length = retrievedMessage.getDataLength();

            byte[] msg = new byte[length];

            retrievedMessage.readFully(msg);

            String sMsg = new String(msg);
            System.out.println(sMsg);

        }catch (RuntimeException e){
            e.printStackTrace();
        }catch (MQException e) {
            if (e.reasonCode != 2033) //没有消息
            {
                e.printStackTrace();
                System.out
                        .println("A WebSphere MQ error occurred : Completion code "
                                + e.completionCode
                                + " Reason Code is "
                                + e.reasonCode);
            }
        } catch (java.io.IOException e) {
            System.out
                    .println("An error occurred whilst to the message buffer "
                            + e);
        }
    }
   
    public void SendMsg(byte[] qByte) {
        try {
            MQMessage qMsg = new MQMessage();
            qMsg.write(qByte);
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            
            qQueue.put(qMsg, pmo);

            System.out.println("The message is sent!");
            System.out.println("\tThe message is " + new String(qByte,"GBK"));
        } catch (MQException e) {
            System.out
                    .println("A WebSphere MQ error occurred : Completion code "
                            + e.completionCode + " Reason Code is "
                            + e.reasonCode);
        } catch (java.io.IOException e) {
            System.out
                    .println("An error occurred whilst to the message buffer "
                            + e);
        }

    }
    /** *//**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        MQTest mqst = new MQTest();        
        mqst.init();
        try {
            mqst.SendMsg(mqst.Msg.getBytes("GBK"));
            //mqst.GetMsg();
        } catch (Exception e) {
            e.printStackTrace();
        }
        mqst.finalizer();
    }

}