MQ系统管理编程概述

2004 年 2 月 01 日

在广大用户使用MQ的过程中，大家比较常用而且比较熟悉的是有关MQI的编程，即利用MQ应用程序接口进行消息的发送和接收，而大家对MQ的系统管理编程可能比较陌生，希望得到更多这方面的知识和技巧，本文将向大家介绍该方面的编程内容和技巧。

在广大用户使用MQ的过程中，大家比较常用而且比较熟悉的是有关MQI的编程，即利用MQ应用程序接口进行消息的发送和接收，以C语言为例，最为常见的是使用MQCONN, MQOPEN, MQPUT/MQGET, MQCLOSE, MQDISC等函数进行队列的读写操作；而大家对MQ的系统管理编程可能比较陌生，希望得到更多这方面的知识和技巧，本文将向大家介绍该方面的编程内容和技巧。

所谓的MQ系统管理编程是指使用MQ提供的编程接口，编制特定的应用程序，来进行对MQ各种对象的监控和管理，如：查询通道的状态、队列的深度等运行时状态，查询队列管理器的属性，队列的属性或通道的属性等静态参数；以及对MQ进行各种操作，如：启动/停止通道，创建/删除各种对象，修改各种对象的属性等。

实际上，MQ为我们提供了多种系统管理的手段，其中包括：

• 利用MQ的图形管理界面进行操作、监控和管理：MQ资源管理器为用户提供了良好的管理界面，从而方便用户对中间件系统的配置和管理工作，通过MQ资源管理器，用户可以定义、修改、删除中间件的各种资源、对象，察看各种对象的属性，监控各种对象的运行状态； 
• 利用MQSC命令，通过命令行的方式对MQ进行操作、监控和管理，尤其值得一提的是MQSC命令在各个平台上都是相同的，在一个有若干平台共存的系统中大大地方便和简化了系统管理的工作。 
• 通过特定的系统管理工具进行监控和管理，MQ不仅自身提供了GUI和文本化监控工具来监测和显示对象的状态,并且能够利用各类系统管理软件对整个网络运行状况实施监控；如Tivoli Manager for MQ就是一个功能强大的综合管理工具：它为控制消息和应用程序框架提供了一个集中管理平台。它捕获和显示MQ的事件，并能发现新的队列管理器以及相应的对象，这些对象可以通过中央控制台进行定义和配置。除此之外，许多第三方厂家也提供有类似的管理工具。 

除了以上这三种管理手段之外，MQ还提供了系统管理的编程接口，通过该接口用户可以编写应用程序从而进行自动化的实时监控及管理。MQ的系统管理接口有两种，即Programmable Command Format（可编程命令格式，简称PCF）和MQ Administration Interface（MQ管理接口，简称MQAI）。下面，我们分别介绍这两种编程接口的原理和使用。

第一部分：WebSphere MQ 可编程命令格式（PCF）

 

WebSphere MQ 可编程命令格式（PCF）命令使得管理任务能编写到应用程序中，在程序中可以创建队列、进程等对象的定义以及更改对象的属性等。

PCF定义了命令和回复消息，应用程序通过这些命令和回复消息实现和队列管理器之间的信息交换，PCF 命令和MQSC 命令具有相同的命令集，所有通过MQSC命令能够实现的功能，通过PCF都可以实现，因此，通过WebSphere MQ的应用程序可以实现对MQ对象的管理包括：队列管理器，进程定义，队列和通道等。PCF命令可以被发送到本地队列管理器的命令队列，也可以被发送到某个远程队列管理器的命令队列，因此，应用程序可以通过一个本地队列管理器集中管理网络中的任何本地和远程管理器。

MQ的远程管理机制底层就是通过PCF这种方式的，在互相联接的系统中的任意一个节点都可以进行对其他所有节点的配置和管理，这种情形的典型应用就是通过一台Windows操作系统的机器来管理全网的MQ节点。由于MQ在Windows XP/NT/2000平台上提供了图形界面的管理工具，我们可以把一个节点设成管理机，利用管理机可以监控和配置网络中的任一节点，监测和显示整个网络中任何一个节点上的服务器及其各种对象的状态和运行情况，从而实现对中间件的集中管理和监控。

每一个队列管理器有一个名为SYSTEM.ADMIN.COMMAND.QUEUE的管理队列，应用程序可以按照PCF命令消息格式封装的要求，组成PCF消息，并将该PCF命令消息发送到管理队列中，同时，每一个队列管理器也有一个命令服务器(Command Server)，它为管理队列中的消息提供服务，在我们使用MQ的控制命令strmqcsv启动命令服务器之后，它将监控管理队列，一旦该队列中有PCF消息到达，它将读取该消息，并解释执行。因此在网络中的任何队列管理器都可以处理PCF消息，通过使用指定的回复队列，回复消息可以被返回给应用程序，应用程序可以获知PCF命令执行成功与否。回复队列由命令消息的消息描述符(MQMD)中的ReplyToQ和ReplyToQMgr两个字段来指定。

PCF命令和回复消息是使用MQ相应的编程接口进行发送和接收的，以C语言为例，我们只需要使用MQPUT将PCF命令消息发送到相应的管理队列，使用MQGET将PCF回复消息从相应的回复队列中取出即可。这里的关键就是如何封装PCF消息。每个MQ指令及其相关参数都是一条单独的命令消息，每个命令消息由PCF头和若干个参数结构块组成，每个参数结构块提供了命令的参数。回复消息的结构与命令消息相同，但是回复消息的个数根据不同的情况可能会有多个，例如：如果我们查询某个队列管理器下所有本地队列的属性，假设本地队列有10个，那么我们将得到10个回复消息，PCF头中的Control字段MQCFC_NOT_LAST和MQCFC_LAST将用于区分是否为最后一个回复消息。

PCF编程接口支持C，Visual Basic, COBOL, RPG, PL/1和Java等，其中在我们最常用的编程语言中，C和Visual Basic编程在PCF的封装上相对Java将会略微复杂一些。

PCF头用来描述希望执行的PCF命令以及紧接PCF头之后的参数结构块的个数，例如：如：MQCMD_INQUIRE_Q 的PCF命令用来查询队列的属性，MQCMD_INQUIRE_CHANNEL_STATUS用来查询通道的运行状态；当希望查询通道的状态时，我们要指出欲查询通道的名称以及类型，这样将会有2个参数结构块。

在PCF头之后紧接着是各个参数结构块，参数结构块有PCF String, PCF Integer, PCF Integer List，PCF String List四种类型。通常情况下，PCF String用来描述PCF命令要操作的对象的名称，例如，若我们要查询名为TEST通道的状态时，我们要将PCF String的通道名称设置为TEST；PCF Integer用来描述PCF命令要操作的对象的类型, 例如，若我们要查询名为TEST通道的当前运行状态时，我们要将PCF Integer中的通道实例类型名称设置为MQOT_CURRENT_CHANNEL； PCF Integer List用来描述PCF命令要操作的对象的某些属性，若我们要查询名为TEST通道的当前运行状态和当前的通道消息序列号时，我们要在PCF List中设置MQIACH_CHANNEL_STATUS和MQCACH_CURRENT_LUWID两个属性。下面，我们给出一个使用C语言来查询通道状态的程序示例片断，见附件1：GetChlStatus.c。

以上我们介绍了PCF的原理和基于C语言的编程实例，下面，我们再给大家介绍一下PCF对Java编程接口的支持。缺省情况下，在您所购买的MQ的产品介质中，是不提供PCF for Java开发环境的，但是，IBM通过SupportPac的形式免费向所有用户提供该支持。相信使用和熟悉MQ的用户，一定知道MQ及其产品家族的SupportPac，各种SupportPac可以供用户从MQ家族产品的网站上免费下载获得，如果登陆http://www.ibm.com/software/mqseries/txppacs，大家就能够获得与之相关的信息，并且可以免费下载其中感兴趣的内容，这里为我们广大用户提供了一些非常有用的工具。例如：在这些支持软件包中，你可以得到有关产品配置方法的信息，有关产品系统管理的信息，有关产品性能测试结果的信息，有关产品规划时的一些建议和方法论，有关产品使用中的一些使用工具，有关产品编程方面的一些样例程序等。

其中，编号为ms0b、名称为"MQSeries Java Classes for PCF"的SupportPac，就提供了PCF的Java编程接口支持。它其中包括了用于PCF Java编程的开发包：com.ibm.mq.pcf.jar、有关PCF Java开发的说明文档以及若干示例程序。

在PCF Java接口中，为我们提供了CMQC, CMQCFC, CMQXC, MQCFH, MQCFIL, MQCFIN, MQCFSL, MQCFST, PCFAgent, PCFMessage, PCFMessageAgent, PCFHeader, PCFParameter, PCFException等若干类，每个类有其相应的属性和方法，通过对这些类的属性设置和方法调用，我们就可以通过Java编写PCF程序，实现对MQ的系统管理编程。

下面，我们仍以查询通道的当前运行状态为例，来介绍PCF Java的基本方法和步骤。首先进行变量的初始化和PCF命令消息的封装：

 PCFAgent   agent;
             int []    attrs = 
                 {CMQCFC.MQIACH_CHANNEL_STATUS};
             PCFParameter []  parameters = 
                 {
                //设置通道名称为TEST，它属于我们前面讲述的PCF String
            new MQCFST (CMQCFC.MQCACH_CHANNEL_NAME,"TEST");, 
                //设置查询的是当前运行的通道，它属于我们前面讲述的PCF Integer
            new MQCFIN (CMQCFC.MQIACH_CHANNEL_INSTANCE_TYPE, CMQC.MQOT_CURRENT_CHANNEL), 
               //设置查询的是当前运行状态属性，它属于我们前面讲述的PCF Integer List
            new MQCFIL (CMQCFC.MQIACH_CHANNEL_INSTANCE_ATTRS, attrs)
                 };
             MQMessage []   responses;
             
                

接下来，创建PCFAgent类，并进行相应的函数调用，发送PCF命令消息：

    if (args.length == 1)
              {
               System.out.print ("Connecting to local queue manager " + 
                 args [0] + "... ");
                //直接连接本地队列管理器
            agent = new PCFAgent (args [0]);
               }
             else
              {
               System.out.print ("Connecting to queue manager at " + 
               args[0] + ":" + args[1] + "over channel" + args[2] + "... ");
               //以MQ 客户端的方式连接队列管理器
            agent = new PCFAgent(args[0], Integer.parseInt(args[1]), args[2]);
               }
               System.out.println ("Connected.");
               agent.setCharacterSet(1381);
            // 通过PCFAgent发送PCF命令
               System.out.print ("Sending PCF request... ");
               //发送的PCF命令为查询通道状态
               responses = agent.send (CMQCFC.MQCMD_INQUIRE_CHANNEL_STATUS, parameters);
               System.out.println ("Received reply.");
               
                

这里，PCFAgent是很重要的一个类，它维护了与队列管理器之间的一个连接，并且提供相应的方法来发送PCF命令消息并获得回复消息。并且，从版本V2.0开始，PCF Classes for Java又为我们提供了名称为PCFMessageAgent的另外一个类，它是PCFAgent的子类，它使用在版本V2.0推出的另外一个新的类：PCFMessage来封装PCF命令消息和回复消息，从而避免了用户对MQ的消息，PCF头以及参数结构块的直接操作，因此，比PCFAgent更加简单易用，而更加受到广大用户的青睐。

为了让您体验一下PCFMessageAgent的方便之处，下面我们给出使用PCFMessageAgent来查询通道当前运行状态的程序示例：

首先进行变量的初始化：

PCFMessageAgent agent;
               PCFMessage  request;
               PCFMessage []  responses;
                

接下来，创建PCFMessageAgent类，并进行相应的函数调用，发送PCF命令消息，：

   if (args.length == 1)
               {
                System.out.print ("Connecting to local queue manager" + 
                 args [0] + "... ");
                //直接连接本地队列管理器
                agent = new PCFMessageAgent (args [0]);
               }
               else
               {
                System.out.print ("Connecting to queue manager at " + 
                 args[0] + ":" + args[1] + "over channel" + args[2] + "... ");
                //以MQ 客户端的方式连接队列管理器
            agent = new PCFMessageAgent (args [0], Integer.parseInt (args [1]), args [2]);   }
               System.out.println ("Connected.");
              // 创建PCF命令消息
              request = new PCFMessage (CMQCFC.MQCMD_INQUIRE_CHANNEL_STATUS);
              request.addParameter (CMQCFC.MQCACH_CHANNEL_NAME,"TEST");
            request.addParameter (CMQCFC.MQIACH_CHANNEL_INSTANCE_TYPE, CMQC.MQOT_CURRENT_CHANNEL);
              request.addParameter (CMQCFC.MQIACH_CHANNEL_INSTANCE_ATTRS,new int [] {CMQCFC.MQIACH_CHANNEL_STATUS});
            // 通过PCFMessageAgent发送PCF命令消息
               System.out.print ("Sending PCF request... ");
               responses = agent.send (request);
               System.out.println ("Received reply.");
               
                

从上面的程序代码可以看出，使用PCFMessageAgent和PCFMessage这两个类，大大简化了对PCF命令消息的封装工作，只需要给出相应的参数即可。我们在附件2中给出了PCFMessageChannelStatus.java程序源代码。

回页首

第二部分：WebSphere MQ 管理接口(MQAI)

 

除了PCF的系统管理编程接口之外，WebSphere MQ还提供另外一种系统管理编程接口，即：MQ管理接口（MQ Administration Interface，简称为MQAI）。MQAI是MQ 提供的一种简化的、实现发送和接收PCF命令消息和回复消息的接口，MQAI通过使用数据包（Data Bags）来处理对象的属性，这样比直接使用PCF更简单。

在MQ for Windows, AIX, iSeries, Linux, HP-UX, and Solaris and WebSphere MQ for OS/2 Warp版本，都支持MQAI。目前，MQAI支持的编程语言不如传统的PCF那么丰富，只有C和Visual Basic两种。

MQAI的底层工作机制同PCF一样，也是通过发送PCF命令消息到MQ命令服务器队列，从而被命令服务器解释执行，并等待回复消息来管理WebSphere MQ，如图所示：

图：MQAI的原理图
 

MQAI通过传递参数到数据包的方法，从而提供了更简单的访问PCF消息的编程接口。这样只要一条语句就可以实现一个结构，编程人员不需要处理数组和分配空间，也不需要了解PCF的消息结构。

每个数据包含有若干个数据项(Data Items)，这些数据项在数据包内被有序排列，它们的排列顺序被称为插入序号(Insertion Order)。每个数据项由一个选择器（Selector）和一个数值（Data Value）组成，其中，选择器是数据项的标识，而数值可以是整数、字符串或者另外一个数据包的句柄。

数据包有用户包(User Bag)， 管理包(Administration Bag)，命令包(Command Bag)和系统包(System Bag)等不同的类型，其中，管理包用于封装发给MQ命令服务器的管理命令，管理包自动包含若干选项；系统包将在命令服务器生成回复消息时自动生成。数据项则有用户数据项和系统数据项两种，其中，用户数据项包含了诸如你要管理的对象属性的数据；系统数据项用于控制生成的消息，诸如消息头的生成。

使用MQAI的一般步骤如下：

1． 确定你要采取的系统管理行为，明确你要执行的命令；

2． 确定选择器和相关的参数；

3． 利用mqCreateBag函数调用生成数据包，并且使用mqAddInteger, mqAddString, mqAddInquiry函数调用生成选择器对应的数据项；

4． 检查MQ命令服务器是否处于运行状态；

5． 使用mqExecute函数调用向MQ命令服务器发送消息，并获得相应的回复消息。

下面，我们以C语言为例，给出使用MQAI的程序样例，这里，我们将使用MQAI来查询队列的当前深度。

首先，进行变量初始化，并建立与队列管理器的连接：

MQHCONN hConn;  /* MQ 连接句柄 */
            MQCHAR qmName [MQ_Q_MGR_NAME_LENGTH+1 ]="";  /*缺省队列管理器*/
            MQLONG reason;   /*原因码*/
            MQLONG connReason;  /*MQCONN reason code */
            MQLONG compCode;  /*完成码*/
            MQHBAG adminBag =MQHB_UNUSABLE_HBAG;  /*mqExecute使用得管理数据包*/
            MQHBAG responseBag =MQHB_UNUSABLE_HBAG;  /*mqExecute的响应数据包*/
            MQHBAG qAttrsBag;  /*含有队列属性的数据包*/
            MQHBAG errorBag;  
            MQLONG mqExecuteCC; /*mqExecute完成码*/
            MQLONG mqExecuteRC; /*mqExecute原因码*/
            MQLONG qNameLength; 
            MQLONG qDepth;   
            MQLONG i;    
            MQLONG numberOfBags; 
            MQCHAR qName [MQ_Q_NAME_LENGTH+1 ];
            //连接队列管理器
            if (argc >1)
            strncpy(qmName,argv [1 ],(size_t)MQ_Q_MGR_NAME_LENGTH);
            MQCONN(qmName,&hConn,&compCode,&connReason);
            if (compCode ==MQCC_FAILED)
            {
            CheckCallResult("Queue Manager connection",compCode,connReason
            exit((int)connReason);
            }
                

第二步，创建用于封装命令消息的管理数据包

mqCreateBag(MQCBO_ADMIN_BAG,&adminBag,&compCode,&reason);
            CheckCallResult("Create admin bag",compCode,reason);
            //创建用于封装回复消息的数据包
            mqCreateBag(MQCBO_ADMIN_BAG,&responseBag,&compCode,&reason);
            CheckCallResult("Create response bag",compCode,reason);
            //向管理数据包中添加要查询的队列的名称，这里，我们将查询所有本地队列的深度
            mqAddString(adminBag,MQCA_Q_NAME,MQBL_NULL_TERMINATED,"*",
            &compCode,&reason);
            CheckCallResult("Add q name",compCode,reason);
            //向管理数据包中添加要查询的队列的类型，这里，我们将查询所有本地队列的深度
            mqAddInteger(adminBag,MQIA_Q_TYPE,MQQT_LOCAL,&compCode,&reason);
            CheckCallResult("Add q type",compCode,reason);
            //向管理数据包中添加要查询的是队列的当前深度这一属性mqAddInquiry(adminBag,MQIA_CURRENT_Q_DEPTH,&compCode,&reason);
            CheckCallResult("Add inquiry",compCode,reason);
            //调用mqExecute，通过该调用将产生PCF消息，然后会将此消息发送到管理命令队列：SYSTEM.ADMIN.CONMMAND.QUEUE，并且得到含有回复消息的应答数据包
            mqExecute(hConn,   /* MQ 连接句柄 */
            MQCMD_INQUIRE_Q,  /*要执行的命令 */
            MQHB_NONE,   
            adminBag,    /*管理数据包句柄 */
            responseBag,    /*响应数据报句柄*/
            MQHO_NONE, /*将消息发送到SYSTEM.ADMIN.COMMAND.QUEUE*/
            MQHO_NONE,   /*为响应产生一个动态队列*/
            &compCode,    /* mqExecute的完成码 */
            &reason);   /* mqExecute call 的原因码*/
                

第三步，处理响应数据包：

if (compCode ==MQCC_OK )/*Successful mqExecute */
            {
            //计算响应数据包中系统数据包的个数，每个队列的属性分别在不同的数据包中
            mqCountItems(responseBag,MQHA_BAG_HANDLE,&numberOfBags,&compCode, &reason);
            CheckCallResult("Count number of bag handles",compCode,reason);
            for (i=0;i<numberOfBags;i++)
            {
            //处理每个系统数据包
            mqInquireBag(responseBag,MQHA_BAG_HANDLE,i,&qAttrsBag,&compCode,
            &reason);
            CheckCallResult("Get the result bag handle",compCode,reason);
            //获得队列名称
            mqInquireString(qAttrsBag,MQCA_Q_NAME,0,MQ_Q_NAME_LENGTH,qName, &qNameLength,NULL,&compCode,&reason);
            CheckCallResult("Get queue name",compCode,reason);
            //获得队列当前深度
            mqInquireInteger(qAttrsBag,MQIA_CURRENT_Q_DEPTH,MQIND_NONE,&qDepth,
            &compCode,&reason);
            CheckCallResult("Get depth",compCode,reason);
            mqTrim(MQ_Q_NAME_LENGTH,qName,qName,&compCode,&reason)
            printf("%4d %-48s \n",qDepth,qName);
            }
            }
            else 
            {
            printf("Call to get queue attributes failed:Completion Code =%d :
            Reason =%d \n",compCode,reason);
            //错误处理，获得错误处理包
            if (reason ==MQRCCF_COMMAND_FAILED)
            {
            mqInquireBag(responseBag,MQHA_BAG_HANDLE,0,&errorBag,&compCode,
            &reason);
            CheckCallResult("Get the result bag handle",compCode,reason);
            //错误处理，获得错误处理包
            mqInquireInteger(errorBag,MQIASY_COMP_CODE,MQIND_NONE,&mqExecuteCC, &compCode,&reason );
            CheckCallResult("Get the completion code from the result bag", compCode,reason);
            //错误处理，从错误处理包中获得原因码
            mqInquireInteger(errorBag,MQIASY_REASON,MQIND_NONE,&mqExecuteRC, &compCode,&reason);
            CheckCallResult("Get the reason code from the result bag", compCode,reason);
            printf("Error returned by the command server:Completion Code =%d : Reason =%d \n",mqExecuteCC,mqExecuteRC);
                

第四步，删除所使用的数据包：

if (adminBag !=MQHB_UNUSABLE_HBAG)
            {
            mqDeleteBag(&adminBag,&compCode,&reason);
            CheckCallResult("Delete the admin bag",compCode,reason);
            }
            if (responseBag !=MQHB_UNUSABLE_HBAG)
            {
            mqDeleteBag(&responseBag,&compCode,&reason);
            CheckCallResult("Delete the response bag",compCode,reason);
            }
                

最后，需要断开与队列管理器的连接：

if (connReason !=MQRC_ALREADY_CONNECTED)
            {
            MQDISC(&hConn,&compCode,&reason);
            CheckCallResult("Disconnect from queue manager",compCode,reason);
            }