ibm-mq - 读取 MQ 分段消息时出现问题

Question

我编写了一个客户端来读取通过 MQ 发送的分段消息，但收到错误消息。下面的代码工作得很好，但它一次只读取一条消息，应用程序必须连接分段的消息，这不是我想要的

public void getMessage(){

        try {

           MQEnvironment.hostname = ""; 
           MQEnvironment.channel = ""; 
           MQEnvironment.port = ;

           MQQueueManager QMgr = new MQQueueManager("MQManager");

           MQGetMessageOptions gmo = new MQGetMessageOptions();

           gmo.options =  MQConstants.MQOO_INPUT_AS_Q_DEF|
                         MQConstants.MQGMO_WAIT|
                         MQConstants.MQGMO_ALL_SEGMENTS_AVAILABLE|
                         MQConstants.MQGMO_LOGICAL_ORDER;

           gmo.matchOptions = MQConstants.MQMO_NONE;
           gmo.waitInterval = MQConstants.MQWI_UNLIMITED;

           MQMessage message = new MQMessage();
           MQQueue queue = QMgr.accessQueue("QName",
                                           gmo.options);

           while(true){

            queue.get(message, gmo);
            int dataLength = message.getDataLength();           
            System.out.println(message.readStringOfCharLength(dataLength));
            message.clearMessage();
           }
        } catch (Exception e) {

            e.printStackTrace();
        }
    }

当我将 MQGMO_SYNCPOINT 添加到选项时，它失败并显示 com.ibm.mq.MQException:MQJE001:完成代码“2”，原因“2046”。

不知道为什么会失败，如果我至少可以让它工作，我将能够在安全地处理应用程序中的所有分段消息后提交。

public void getMessage(){

        try {

           MQEnvironment.hostname = ""; 
           MQEnvironment.channel = ""; 
           MQEnvironment.port = ;

           MQQueueManager QMgr = new MQQueueManager("MQManager");

           MQGetMessageOptions gmo = new MQGetMessageOptions();

           gmo.options =  MQConstants.MQOO_INPUT_AS_Q_DEF|
                         MQConstants.MQGMO_WAIT|
                         MQConstants.MQGMO_ALL_SEGMENTS_AVAILABLE|
                         MQConstants.MQGMO_LOGICAL_ORDER|
                         MQConstants.MQGMO_SYNCPOINT;

           gmo.matchOptions = MQConstants.MQMO_NONE;
           gmo.waitInterval = MQConstants.MQWI_UNLIMITED;

           MQMessage message = new MQMessage();
           MQQueue queue = QMgr.accessQueue("QName",
                                           gmo.options);

           while(true){

               queue.get(message, gmo);
                int dataLength = message.getDataLength();           
                System.out.println(message.readStringOfCharLength(dataLength));
                QMgr.commit();
                message.clearMessage();
           }
        } catch (Exception e) {

            e.printStackTrace();
        }
    }

当我尝试将分段消息作为单个消息阅读时；它失败了com.ibm.mq.MQException:MQJE001:完成代码“2”，原因“2046”。

如果有人能帮助解决这个问题，我将不胜感激；我不太确定下面的代码有什么问题。这是我阅读分段消息的首选方式。

public void getMessage(){


        try {

           MQEnvironment.hostname = ""; 
           MQEnvironment.channel = ""; 
           MQEnvironment.port = ;

           MQQueueManager QMgr = new MQQueueManager("MQManager");

           MQGetMessageOptions gmo = new MQGetMessageOptions();

           gmo.options =  MQConstants.MQOO_INPUT_AS_Q_DEF|
                         MQConstants.MQGMO_WAIT|
                         MQConstants.MQGMO_COMPLETE_MSG;

           gmo.matchOptions = MQConstants.MQMO_NONE;
           gmo.waitInterval = MQConstants.MQWI_UNLIMITED;

           MQMessage message = new MQMessage();
           MQQueue queue = QMgr.accessQueue("QName",
                                           gmo.options);

           while(true){

               queue.get(message, gmo);
                int dataLength = message.getDataLength();           
                System.out.println(message.readStringOfCharLength(dataLength));
               message.clearMessage();
           }
        } catch (Exception e) {

            e.printStackTrace();
        }
    }

Answer 1

问题是您正在使用 gmo.options 作为 accessQueue 方法的打开选项和 get 的获取消息选项.您正在将打开选项和获取消息选项混合到该字段中。

每个选项都由选项字段中的一位表示。当您将该字段用作打开选项时，即使您指定了获取消息选项，这些位也会被 MQ 解释为打开选项，相反，当您将该字段用作获取消息选项时，它们将被 MQ 解释为获取消息选项。

---

让我们看看您在示例中指定的每个选项的值(以十六进制表示)以及基于该值的等效打开或获取选项:

MQOO_INPUT_AS_Q_DEF            0x00000001
MQGMO_WAIT                     0x00000001

MQOO_INPUT_SHARED              0x00000002
MQGMO_SYNCPOINT                0x00000002

MQOO_RESOLVE_NAMES             0x00010000
MQGMO_COMPLETE_MSG             0x00010000

MQOO_BIND_NOT_FIXED            0x00008000
MQGMO_LOGICAL_ORDER            0x00008000

MQOO_RESOLVE_LOCAL_Q           0x00040000
MQOO_RESOLVE_LOCAL_TOPIC       0x00040000
MQGMO_ALL_SEGMENTS_AVAILABLE   0x00040000

因为您错误地将其用作 QMgr.accessQueue("QName",gmo.options) 的打开选项； MQ 以意想不到的方式对此进行解释。

---

对于您的第一个“工作”示例，打开选项将按以下指定进行解释，这些特定的打开选项一起不会对本地队列造成任何问题，这就是为什么它“工作”，即使它不正确:

MQOO_INPUT_AS_Q_DEF            0x00000001
MQOO_INPUT_AS_Q_DEF            0x00000001 //MQGMO_WAIT
MQOO_RESOLVE_LOCAL_Q           0x00040000 //MQGMO_ALL_SEGMENTS_AVAILABLE
MQOO_RESOLVE_LOCAL_TOPIC       0x00040000 //MQGMO_ALL_SEGMENTS_AVAILABLE
MQOO_BIND_NOT_FIXED            0x00008000 //MQGMO_LOGICAL_ORDER

---

在您的第二个示例中，2046 (MQRC_OPTIONS_ERROR) 失败，这是因为选项正在按以下指定进行解释。您不能在打开的选项中同时使用 MQOO_INPUT_AS_Q_DEF 和 MQOO_INPUT_SHARED，这会导致 2046:

MQOO_INPUT_AS_Q_DEF            0x00000001
MQOO_INPUT_AS_Q_DEF            0x00000001 //MQGMO_WAIT
MQOO_RESOLVE_LOCAL_Q           0x00040000 //MQGMO_ALL_SEGMENTS_AVAILABLE
MQOO_RESOLVE_LOCAL_TOPIC       0x00040000 //MQGMO_ALL_SEGMENTS_AVAILABLE
MQOO_BIND_NOT_FIXED            0x00008000 //MQGMO_LOGICAL_ORDER
MQOO_INPUT_SHARED              0x00000002 //MQGMO_SYNCPOINT

---

在您的第三个示例中，2046 (MQRC_OPTIONS_ERROR) 失败，这是因为选项的解释如下所述。 MQOO_RESOLVE_NAMES 被记录为仅在 MQ C++ API 中有效，这会导致 2046:

MQOO_INPUT_AS_Q_DEF            0x00000001
MQOO_INPUT_AS_Q_DEF            0x00000001 //MQGMO_WAIT
MQOO_RESOLVE_NAMES             0x00010000 //MQGMO_COMPLETE_MSG

---

使用 MQOO_INPUT_AS_Q_DEF 作为获取消息选项不会导致任何问题，因为它与 MQGMO_WAIT 具有相同的值，您已经在每个示例中拥有了这个不会改变获取消息选项的行为。

MQGMO_WAIT                     0x00000001 //MQOO_INPUT_AS_Q_DEF
MQGMO_WAIT                     0x00000001

---

基于您的第二个和第三个示例的以下内容应该有效:

public void getMessage(){


        try {

           MQEnvironment.hostname = ""; 
           MQEnvironment.channel = ""; 
           MQEnvironment.port = ;

           MQQueueManager QMgr = new MQQueueManager("MQManager");

           // Set up the options on the queue we wish to open
           int openOptions = MQConstants.MQOO_INPUT_AS_Q_DEF;

           MQGetMessageOptions gmo = new MQGetMessageOptions();

           gmo.options =  MQConstants.MQGMO_WAIT|
                          MQConstants.MQGMO_ALL_SEGMENTS_AVAILABLE|
                          MQConstants.MQGMO_LOGICAL_ORDER|
                          MQConstants.MQGMO_SYNCPOINT|
                          MQConstants.MQGMO_COMPLETE_MSG;

           gmo.matchOptions = MQConstants.MQMO_NONE;
           gmo.waitInterval = MQConstants.MQWI_UNLIMITED;

           MQMessage message = new MQMessage();
           MQQueue queue = QMgr.accessQueue("QName", openOptions);

           while(true){

               queue.get(message, gmo);
               int dataLength = message.getDataLength();           
               System.out.println(message.readStringOfCharLength(dataLength));
               QMgr.commit();
               message.clearMessage();
           }
        } catch (Exception e) {

            e.printStackTrace();
        }
    }

---

如果您运行完整 MQ 客户端安装附带的 mqrc 实用程序，您可以找出错误代码的含义:

$mqrc 2046

      2046  0x000007fe  MQRC_OPTIONS_ERROR