Java调用MQ队列

Posted 2020-10-07 申公的博客

转载：http://blog.csdn.net/ozwarld/article/details/7735915

IBM MQ 6.0中设置两个队列，（远程队列、通道之类都不设置）。

队列管理器是XIR_QM_1502

队列名称是ESBREQ

IP地址是10.23.117.134（远程的一台电脑，跟我的电脑不在一个局域网内）

端口1414

CCSID 1208

 

MQ配置可以参考这个，有配图http://wenku.baidu.com/view/06d108d0360cba1aa811daa3.html

程序如下，发送线程两个，接收线程一个。接收完毕后就结束。

/* 
 * 创建日期 2012-7-10 
 * 
 * TODO 要更改此生成的文件的模板，请转至 
 * 窗口 － 首选项 － Java － 代码样式 － 代码模板 
 */  
package yerasel;  
  
/** 
 * @author Fenglb E-mail:[email protected] 
 * @version 创建时间：2009-4-30 下午04:13:38 类说明 
 */  
  
import java.io.IOException;  
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;  
  
interface SomeConstants {  
    String qManager = "XIR_QM_1502";//"XIR_QM"; //QueueManager name  
    String qName = "ESBREQ";// Queue Name  
    String strIP = "10.23.117.134";//"10.24.28.139";//"10.24.28.102";  
    int iPort = 1502;//1414;  
    String strChl = "SYSTEM.DEF.SVRCONN";// Server-Connection Channel  
    int iCCSID = 1208;  
}  
  
class Sender implements Runnable, SomeConstants {  
    public void run() {  
        sendMessage();  
    }  
  
    public void sendMessage() {  
  
        String name = Thread.currentThread().getName();  
        System.out.println("进入线程" + name);  
  
        MQQueueManager qMgr = null;  
        // configure connection parameters  
  
        MQEnvironment.hostname = strIP;  
        // Server name or IP  
        MQEnvironment.port = iPort;  
        MQEnvironment.channel = strChl;  
        MQEnvironment.CCSID = iCCSID;  
  
        // java程序连接mq的方式有两种，一是客户机方式，一是绑定方式，  
        // 默认是客户机方式，当mq部署在本地的时候，就需要用绑定方式  
        // 本机IP是10.24.28.139连接10.23.117.134的时候不需要下句  
        //MQEnvironment.properties.put(MQC.TRANSPORT_PROPERTY,  
        //MQC.TRANSPORT_MQSERIES_BINDINGS);  
  
        // Create a connection to the QueueManager  
        System.out.println(name + " Connecting to queue manager: " + qManager);  
        try {  
            qMgr = new MQQueueManager(qManager);  
            // Set up the options on the queue we wish to open  
            int openOptions = MQC.MQMT_REQUEST | MQC.MQPMO_NEW_MSG_ID  
                    | MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING  
                    | MQC.MQOO_INPUT_AS_Q_DEF;  
            // Now specify the queue that we wish to open and the open options  
            System.out.println(name + " Accessing queue: " + qName);  
            MQQueue queue = qMgr.accessQueue(qName, openOptions);  
            // Define a simple WebSphere MQ Message ...  
  
            // Specify the default put message options  
            MQPutMessageOptions pmo = new MQPutMessageOptions();  
  
            // Put the message to the queue  
            System.out.println(name + " Sending a message...");  
  
            MQMessage msg = new MQMessage();  
            msg.messageId = "MSGID".getBytes();  
            msg.messageType = MQC.MQMT_REQUEST;  
            msg.replyToQueueName = "ESBREQ";  
  
            // 在此测试一下 mq 的传输次列  
            for (int j = 1; j < 5; j++) {  
                msg.messageSequenceNumber = j;  
                // write some text in UTF8 format  
                try {  
                    String str = "Salemetsizbe Yerasel";  
                    str = str + " " + j;  
                    msg.writeUTF(str);  
                    queue.put(msg, pmo);  
                    msg.clearMessage();  
                    System.out.println(name + " putting the message... " + j);  
                } catch (MQException mqe) {  
                    mqe.printStackTrace();  
                    break;  
                } catch (IOException e1) {  
                    e1.printStackTrace();  
                }  
            }  
            qMgr.commit();  
            System.out.println(name + " Done!");  
            System.out.println("==========");  
            System.out.println("");  
        } catch (MQException e) {  
            e.printStackTrace();  
        }  
    }  
}  
  
class Receiver implements Runnable, SomeConstants {  
  
    public void run() {  
        recvMessage();  
    }  
  
    public void recvMessage() {  
  
        String name = Thread.currentThread().getName();  
          
        try {  
            Thread.sleep(1000);  
            MQQueueManager qMgr = null;  
  
              
            System.out.println("进入线程" + name);  
  
            System.out.println(name + " Connecting to queue manager: "  
                    + qManager);  
            qMgr = new MQQueueManager(qManager);  
            // 设置将要连接的队列属性  
            // 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;  
  
            // Now get the message back again. First define a WebSphere MQ  
            // message to receive the data  
            MQMessage rcvMessage = new MQMessage();  
  
            // Specify default get 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 = 1000; // Sets the time limit for the  
                                        // wait.（设置等待的毫秒时间限制）  
  
            System.out.println(name + " Accessing queue: " + qName);  
            MQQueue queue = qMgr.accessQueue(qName, openOptions);  
            int depth = 0;  
  
            // Get the message off the queue.  
            System.out.println("... " + name + " getting the message back again");  
            for (;;) {  
                try {  
                    queue.get(rcvMessage, gmo);  
                    System.out.println(" ID: "  
                            + (new String(rcvMessage.messageId)).trim()  
                            + " Num: " + rcvMessage.messageSequenceNumber  
                            + " Type: " + rcvMessage.messageType + " Flag: "  
                            + rcvMessage.messageFlags);  
                    // And display the message text...  
                    String msgText = rcvMessage.readUTF();  
                    System.out.println("The message is: " + msgText);  
                    rcvMessage.clearMessage();  
  
                    // Break if no MSG left in queue  
                    depth = queue.getCurrentDepth();  
                    if (depth == 0)  
                        break;  
  
                } catch (MQException mqe) {  
                    mqe.printStackTrace();  
                    break;  
                    // null;  
                } catch (IOException e) {  
                    e.printStackTrace();  
                }  
            }  
            // Close the queue  
            System.out.println(name + " Closing the queue");  
            queue.close();  
            // Disconnect from the QueueManager  
            System.out.println(name + " Disconnecting from the Queue Manager");  
            qMgr.disconnect();  
            System.out.println(name + " Done!");  
            System.out.println("==========");  
            System.out.println("");  
        } catch (MQException ex) {  
            System.out  
                    .println("A WebSphere MQ Error occured : Completion Code "  
                            + ex.completionCode + " Reason Code "  
                            + ex.reasonCode + ex.getMessage());  
        } catch (InterruptedException e1) {  
            e1.printStackTrace();  
        }  
    }  
}  
  
public class MQTest {  
  
    public static void main(String args[]) {  
  
        /* 
         * MQTest first = new MQTest(); first.sendMessage(); 
         * first.recvMessage(); 
         */  
        Sender sender = new Sender();  
        Thread senderThread = new Thread(sender);  
        senderThread.start();  
        senderThread.setName("Sender");  
          
        Thread senderThread2 = new Thread(sender);  
        senderThread2.start();  
        senderThread2.setName("Sender2");  
          
        Receiver recv = new Receiver();  
        Thread recvThread = new Thread(recv);  
        recvThread.start();  
        recvThread.setName("Receiver");  
  
        // Receiver recv = new Receiver();  
        // new Thread(recv).start();  
  
    }  
  
}