c - WebSphere 非托管队列 MQSUB 返回 2019 错误

Question

我想知道是否有人是非托管 WebSpehre 消息队列方面的专家可以帮助我。我们有一个应用程序在构造函数中执行以下操作: 1. MQCONN 到队列管理器。 2. MQOPEN 打开非托管队列。 3. MQSUB 订阅主题列表。在 MQSUB 中，我传入的选项是:“MQSO_CREATE | MQSO_RESUME | MQSO_DURABLE | MQSO_FAIL_IF_QUIESCING”。问题在于，在主题列表中，其中一些主题会失败 MQSUB 调用并返回错误 2019，而其他主题则会成功。我对WebSphere队列不是很熟悉，但是根据我读过的文章，这个错误意味着传入的Hobj与原始MQSUB句柄不匹配。我们尝试重置队列，之后我们看到所有订阅都成功了。然而，在随后的测试中，我们又看到了同样的问题。让我困惑的是，如果Hobj与原始句柄不匹配(因为应用程序没有干净退出，所以订阅没有被清理)，为什么有些会成功？它们不共享从 MQOPEN 调用返回的相同 Hobj 句柄吗？处理这种情况的正确方法是什么？我是否应该在 MQOPEN 调用之后将 Hobj 句柄保存在配置文件中，并且当应用程序再次运行时，从配置文件中读取并传入相同的 Hobj 句柄？或者有一种方法可以在再次运行 MQSUB 之前重置订阅？这会导致消息丢失吗？非常感谢您的见解!

这是代码:

WebSphere_MQ_FrontEnd::WebSphere_MQ_FrontEnd(const CI_String& qMgr, // the queue manager of interested topics
                                               const CI_String& connectionName, // topic list
                                               const CI_String& qName, // the unmanaged queue that will hold the subscribed messages
                                                int debug  // debug flag. Default set to be off.
                                            ) 
    // new constructor
    // Add Debug flag
{
/*    tibrv_status status;
*/
   const char TOPIC_DELIM_DEFAULT = '|';

   /*   Declare MQI structures needed                                */
   MQOD     od = {MQOD_DEFAULT};    /* Object Descriptor             */
      /** note, sample uses defaults where it can **/

   char topicsString[1024]        = "";


   //MQHCONN  Hcon;                   /* connection handle             */
   //MQHOBJ   Hobj;                   /* object handle                 */
   MQSD sd = {MQSD_DEFAULT};       /* Subscription Descriptor */
   MQMD md = {MQMD_DEFAULT};       /* Message Descriptor */
   MQOD     td = {MQOD_DEFAULT};         /* Object descriptor                      */

   //MQGMO gmo = {MQGMO_DEFAULT};    /* get message options */
   MQLONG   O_options;              /* MQOPEN options                */
   MQLONG   C_options;              /* MQCLOSE options               */
   MQLONG   CompCode = 0;           /* completion code               */
   //MQLONG   OpenCode;               /* MQOPEN completion code        */

   MQLONG   Reason;                 /* reason code                   */
   MQLONG   OReason = 0;                 /* reason code for MQOPEN                  */
   MQLONG   CReason;                /* reason code for MQCONN        */

   MQLONG   O_CompCode = 0;             /* newton MQSUB completion code         */ 
   MQLONG   S_CompCode;             /* newton MQSUB completion code         */
   MQBYTE   buffer[65536];            /* message buffer                */
   MQLONG   buflen;                 /* buffer length                 */
   MQLONG   messlen;                /* message length received       */

   MQLONG   SubscribeReason[MAX_SUBSCRIPTIONS] = { -1 };   /* reason code */
   /* reason code */
   char     QMgr[50];               /* queue manager name            */
   char     ReasonString[10];       /* String for Reason Code           */
   MQCNO    ConnectOptions = { MQCNO_DEFAULT };

   ConnectOptions.Options = MQCNO_HANDLE_SHARE_BLOCK;

   int      subidx = 0;

   char     subname[256];
   char     topicname[256];

   //char *   topicString = topicStringDefault;
   _debug = debug;
   _Hcon = MQHC_UNUSABLE_HCONN; /* connection handle                */
   _Hobj = MQHO_NONE;            /* subscription queue handle        */

   std::cout.setf(std::ios::unitbuf);
   if (debug) {
      //printf("Constructor start\n"); 
      std::cout <<  "Constructor start" << '\n';
   }


   /******************************************************************/
   /*                                                                */
   /*   Connect to queue manager                                     */
   /*                                                                */
   /******************************************************************/

   if ((qMgr.size()) < 0) {
        CI_ExError ex(MQ_EMISSINGARG, "qMgr");
      ex.chain(CI_EINVALID, qMgr.data()).throwThis();

   }

   // prepare the subscription topic list. passed in from connectionName argument
      if ( connectionName.size() <0 ) {  // is this the right way to handle the error?
        if (debug) {
            //printf("connection Name size < 0\n");
            std::cout << "connection Name size < 0" << '\n';
        }
        CI_ExError ex(MQ_EMISSINGARG, "connectionName");
      ex.chain(CI_EINVALID, connectionName.data()).throwThis();
    }

   strncpy(topicsString, connectionName.data(), 1024); // how many topics are we expecting?
   //_TopicStringList = NULL;
   str_split(topicsString, TOPIC_DELIM_DEFAULT, _TopicStringList);

    try {
      strncpy(QMgr, qMgr.data(), 50);

      if ( debug) {
        //printf("calling MQCONNX\n"); 
        //printf("    Manager: %s\n" , QMgr); 
        std::cout << "calling MQCONNX" << '\n';
        std::cout << "    Manager: " << QMgr << '\n';
      }
        //printf("    1Options: %s\n" , ConnectOptions.Options);
    /*  removed  printf("    2Options: %ld \n" , CompCode); */


      MQCONN(QMgr,                    /* queue manager                  */
            &_Hcon,                   /* connection handle              */
            &CompCode,               /* completion code                */
            &CReason);               /* reason code                    */




    /* report reason and stop if it failed     */
    if (CompCode == MQCC_FAILED)    {
        //printf("3MQCONN ended with reason code %d\n", CReason);
        std::cout << "3MQCONN ended with reason code " << CReason << '\n';
        sprintf(ReasonString, "%d",CReason);
        CI_ExError ex(MQ_CONNFAILED);
        ex.chain(MQ_REASON, ReasonString).throwThis();
    }
   } catch (CI_ExError& ex) {
         ex.chain(MQ_STOP).throwThis();
   }

   if ( debug ) {
        if (CompCode != MQCC_FAILED)
            //printf("CompCode in MQCONN is OK.\n"  );
            std::cout << "CompCode in MQCONN is OK." << '\n';

        //printf("passed MQCONN\n");
        std::cout << "passed MQCONN" << '\n';
    }
   /******************************************************************/
   /*                                                                */
   /*   Use parameter as the name of the target queue                */
   /*                                                                */
   /******************************************************************/
   if ( debug ) {
        //printf("Checking qName %s is of size qName: \n",qName);
        std::cout << "Checking qName " << qName << " is of size qName:" << '\n'; 
   }
   if ((qName.size()) == 0) {
        CI_ExError ex(MQ_EMISSINGARG, "qName");
      ex.chain(CI_EINVALID, qName.data()).throwThis();
    }
    try {
        // try to open this unmanaged queue passed by user
        strncpy(od.ObjectName, qName.data(), (size_t)MQ_Q_NAME_LENGTH);
        /*  removed   printf("target queue is %s\n", od.ObjectName); */
        MQLONG O_options = MQOO_INPUT_AS_Q_DEF | MQOO_INQUIRE | MQOO_FAIL_IF_QUIESCING;
        if (debug ) {
        std::cout << "od: " << od.ObjectName << " O_options: " << O_options << " OReason:" << OReason<< '\n'; 
        }
        MQOPEN(_Hcon, &od, O_options, &_Hobj, &O_CompCode, &OReason); 
        if (O_CompCode != MQCC_OK)
        {
            //printf("MQOPEN ended with reason code %d\n", OReason);
            std::cout << "MQOPEN ended with reason code " << OReason << '\n';
        }

        if( debug ) {
            if (O_CompCode != MQCC_FAILED) {
                //printf("CompCode in MQOPEN is OK.\n"  );
                std::cout << "CompCode in MQOPEN is OK. Hobj: " << _Hobj << "\n";
            }

            //printf("passed MQOPEN\n");
            std::cout << "passed MQOPEN\n";

        }


    } catch (CI_ExError& ex) {
        /*  removed  cout << "called closed" << endl; */
         close();
         /*  removed  cout << "return from  closed" << endl;*/
         ex.chain(MQ_STOP).throwThis();
    }

    try {
        for (subidx = 0; *(_TopicStringList+subidx); subidx++) {
            sd.Options = MQSO_CREATE | MQSO_RESUME | MQSO_DURABLE | MQSO_FAIL_IF_QUIESCING;
            if (subidx > 0) 
            {
                sd.Options |= MQSO_SET_CORREL_ID;
                memcpy(sd.SubCorrelId, md.CorrelId, MQ_CORREL_ID_LENGTH);
            }

            sprintf(subname, "%s", _TopicStringList[subidx]);
            sprintf( topicname, "%s", strstr( subname, ":") + 1 );
            sd.ObjectString.VSPtr = topicname;
            sd.ObjectString.VSLength = MQVS_NULL_TERMINATED;
            sd.SubName.VSPtr = subname;
            sd.SubName.VSLength = MQVS_NULL_TERMINATED;
            if ( debug ) 
                //printf("Subscribe Topic String[%d]: %s\n", subidx, sd.ObjectString.VSPtr); 
                std::cout << "Subscribe Topic String[" << subidx << "]: \n subname:" << subname <<  " \n topic:" << topicname <<'\n';
            _Hsub[subidx] = MQSO_NONE;  

            MQSUB(_Hcon, &sd, &_Hobj, &_Hsub[subidx], &CompCode, &SubscribeReason[subidx]);
            if (CompCode != MQCC_OK)    { 
                //printf("MQSUB(%d) ended with reason code %d\n", subidx, SubscribeReason[subidx]);
                std::cout << "MQSUB(" << subidx << ") ended with reason code " << SubscribeReason[subidx] << '\n';

                }
            if (debug && CompCode == MQCC_OK)
                // printf( "passed MQSUB [%d]\n", subidx );
                std::cout << "passed MQSUB [" << subidx << "]\n";
            memcpy(md.CorrelId, sd.SubCorrelId, MQ_CORREL_ID_LENGTH);
        }
    } catch (CI_ExError& ex) {
        /*  removed  cout << "called closed" << endl; */
         close();
         /*  removed  cout << "return from  closed" << endl;*/
         ex.chain(MQ_STOP).throwThis();
   }
    if ( debug ) 
        //printf( "this is the End of Constructor\n" );
        std::cout << "this is the End of Constructor\n";
}

Answer 1

MQSUB 返回 2019 的原因之一是应用程序传递的 Hobj 与第一个 MQSUB 调用中提供的队列不同。从知识中心粘贴:

如果提供了 Hobj，它必须与原始 MQSUB 调用中的 Hobj 等效。这意味着，如果提供从 MQOPEN 调用返回的对象句柄，则该句柄必须与之前使用的队列相同。如果不是同一个队列，则调用失败并返回 MQRC_HOBJ_ERROR。

你能检查一下你是否传递了正确队列的Hobj吗？