Основы программирования для WebSphere MQ

Технологические вопросы интеграции приложений

Задачи интеграции приложений возникают достаточно часто в современных корпоративных системах. В качестве примера можно привести такую задачу. В московском динамично развивающемся банке принято решение о переходе с наиболее популярной в России банковской системы компании "Диасофт" на западную банковскую систему T24 компании TEMENOS, получившую широкое распространение в мире и в России за последние годы. Причины, побудившие к этому переходу, могут быть следующие:

В этой задаче для нас интересна технология такого перехода. Совершенно очевидно, что переход от одной автоматизированной банковской системы (АБС) к другой не может пройти за один день или даже за один месяц. Этот переход будет идти несколько месяцев, а возможно, один или два года. В первую очередь это зависит от используемых технологий и системных интеграторов. Кроме этого, должен быть обучен персонал, а сам переход должен быть тщательно протестирован и осуществляться по подсистемам. Проект перехода на новую АБС составляется по подсистемам, в каждой из которых выделяются свои группы задач.

В качестве конкретной задачи рассмотрим создание интерфейса по передаче клиентов из системы "Диасофт" (АБС1) в систему T24 (АБС2). АБС1 функционирует под Windows на основе базы данных SQL Server. В АБС2 предполагается функционирование под UNIX (HP_UX) на основе базы данных ORACLE. Известна структуры данных (таблицы client) в АБС1 и АБС2. Требуется создать интерфейс: клиенты АБС1 => WebSphere MQ => клиенты АБС2. WebSphere MQ идеально подходит для решения такого класса задач по межплатформенной передаче данных, являясь мировым лидером среди транспортных систем.

Существует разные варианты решения поставленной задачи:

Первый путь ясен в технической реализации. С одной стороны при каждом обновлении таблицы client в АБС1 программа-обработчик срабатывает как триггер базы данных и помещает результаты оператора update в очередь, они приходят на АБС2, где своя программа-обработчик срабатывает как триггер очереди и помещает сообщение в таблицу client АБС2. WebSphere MQ гарантирует доставку сообщений. Разработчикам приложений (программ-обработчиков) остается позаботиться 1) о преобразовании форматов АБС1 в АБС2 в одной из программ-обработчиков, например, на платформе Windows; 2) о надежности такой передачи с учетом механизмов транзакционности в WebSphere MQ и в базах данных одновременно. Ведь в банковских системах ничего не должно и не может пропасть! Укрупненная блок-схема программы-обработчика в АБС2 для такого надежного транзакционного взаимодействия выглядит следующим образом.

Блок 1 MQCONN;  MQOPEN;
	Блок 2 MQBEGIN;  MQGET;
	Блок 3 Begin Tran
	Блок 4 UPDATE CLIENT SET ...
	   WHERE ...
	Блок 5 If Error = 0  then
	  Commit Tran
	    else
	  Rollback Tran;
	Блок 6 If Error = 0  then
	  MQCMIT  else MQBACK;
	Блок 7 MQCLOSE;  MQDISC;

В этой программе транзакция WebSphere MQ является внешней по отношению к транзакции базы данных. В программе-обработчике для АБС1 наоборот транзакция WebSphere MQ будет внутренней по отношению к транзакции базы данных. Таким образом, реализация интерфейса по первому варианту не вызывает технических проблем и в следующем разделе мы рассмотрим пример на программирование транзакций для WebSphere MQ. Остается отметить один важный момент. Первый вариант не перспективен при создании нескольких десятков интерфейсов и более. Во-первых, затраты на разработку возрастут по сравнению со вторым и третьим вариантами, когда используются специализированные средства. Во-вторых, сопровождение нескольких десятков разных программ, написанных разными программистами, становится весьма серьезной задачей, а их модификация после увольнения авторов программ или прекращения с ними договорных отношений может оказаться неразрешимой проблемой. К сожалению, жизнь такова, что программисты увольняются, программы интерфейсов живут по несколько лет и их требуется модифицировать. Поэтому нужно очень серьезно подумать в самом начале интеграционного проекта, какой выбрать путь для реализации интерфейсов. Если будет создано несколько интерфейсов по варианту 1, то переделывать их по варианту 2 – задача малоприятная и неблагодарная.

Тем не менее, при небольшом числе интерфейсов рекомендуется вариант 1, как наиболее экономический, к рассмотрению которого мы и переходим. WebSphere BI Broker будет рассмотрен в лекции 12. К сожалению, рассмотрение средств интеграции с WebSphere MQ в системе T24 (вариант 3) выходит за пределы данного курса лекций.

Программирование транзакций

Сообщения WebSphere MQ могут быть четырех типов:

Наша очередная задача: на сервере 1 прочитать сообщение из входной очереди, положить её в очередь для отправки на сервер 2 как сообщение-запрос и дождаться прихода сообщения-ответа, как это показано на рис.9.2. Все это необходимо оформить в виде транзакции, для которой будет осуществляться откат в случае неполучения сообщения-ответа в течении 10 сек. Эта задача может использоваться в практических целях при нестабильной работе каналов, например выделенных. Наше приложение при откате транзакции может попытаться перенаправить сообщений из входной очереди – но это уже другая задача.

Итак, последовательность псевдокода представляется следующим образом (обратите внимание на блок 5 и опции MQMD):

Блок 1 MQCONN
	Блок 2 MQOPEN
	Блок 3 MQBEGIN
	Блок 4 MQGET (Input_queue)
	Блок  5 MQPUT (Output_queue, 
	    MQMD.MsgType = MQMT_REQUEST,
	    MQMD.ReplyToQ = Reply_queue)
	Блок 6 MQGET (Reply_queue)
	Блок 7 If Reply time < 10 sec
	       then
	         MQCMIT  else MQBACK;
	Блок 8 MQCLOSE
	Блок 9 MQDISC

Назовем нашу программу transmit.exe и файл инициализации transmit.ini, в котором 1-я строка – имя очереди для чтения, 2-я строка – имя очереди для записи, 3-я строка – имя очереди для ответа, 4-я строка – время ожидания ответа Reply_time = 3000мсек, как показано ниже.

QUEUE_INPUT
QUEUE_OUTPUT
QUEUE_REPLY
3000

Тип очереди Output_queue – remote queue и эта очередь настроена для отправки сообщений на сервер 2. На сервере 2 также выполнены соответствующие настройки и при нормальной работе каналов транзакция будет совершаться успешно. Отметим также, что сообщение-ответ формируется на сервере 2 средствами другого приложения на этом сервере. В случае остановки любого канала, которую мы произведем для отладки программы, будет происходить откат транзакции. В данной версии в начале программы производится извлечение параметров из ini-файла. Такую программу полезно также иметь в виде триггера и читателю предлагается самостоятельно модифицировать программу для считывания параметров триггера из очереди, на которую он навешивается.

Ниже приводится листинг программы transmit.cpp для Microsoft Visual C++ ver.6.0. Для каждого сообщения MsgId и the CorrelId создаются как уникальные ( MSGID= MQMI_NONE и CORRELID= MQCI_NONE ) и об этом подробнее в лекции 11.

/* Листинг программы transmit */
#include <stdio.h>
 #include <stdlib.h>
 #include <string.h> 
 #include <time.h>
 #include <io.h>     
 #include <cmqc.h>
 char queue_input[48] = "";
 char queue_output[48] = ""; 
 char queue_reply[48] = ""; 
 char reply_time[48] = ""; 
 char buf[48];  
 time_t tmr;
 int queuenamelen;
 FILE *fptr;	

 int main(int argc, char **argv)
 {
   MQOD    odG = {MQOD_DEFAULT};    
   MQOD    odP = {MQOD_DEFAULT};    
   MQOD    odR = {MQOD_DEFAULT};    
   MQOD    odI = {MQOD_DEFAULT};    
   MQMD    md  = {MQMD_DEFAULT};    
   MQBO	   mbo = {MQBO_DEFAULT};
   MQGMO   gmo = {MQGMO_DEFAULT};   
   MQPMO   pmo = {MQPMO_DEFAULT};         
   MQCHAR48 QManager;               
   MQHCONN  Hcon;                   
   MQHOBJ   Hobj;                   
   MQHOBJ   Hout;               
   MQHOBJ   Hrep;               
   MQLONG   O_options;              
   MQLONG   C_options;              
   MQLONG   CompCode;               
   MQLONG   Reason;                 
   MQLONG   CReason;                
   MQBYTE   buffer[8001];           
   MQLONG   buflen;                 
   MQLONG   replylen;   
   MQLONG   messlen;                
 static  MQBYTE24 LastMsgId;				   
  
	if ( (fptr=fopen ("transmit.ini","r" )) == NULL )
		{printf("Cannot open transmit.ini file" ); 
		 exit(1);	
		}
	else{
		fgets(queue_input, 48, fptr);			
		queuenamelen = strlen(queue_input) - 1;						
		queue_input[queuenamelen] = ' ';		
		strcpy(odG.ObjectName, queue_input);

		fgets(queue_output, 48, fptr);			
		queuenamelen = strlen(queue_output) - 1;						
		queue_output[queuenamelen] = ' ';		
		strcpy(odP.ObjectName, queue_output);		

		fgets(queue_reply, 48, fptr);			
		queuenamelen = strlen(queue_reply) - 1;						
		queue_reply[queuenamelen] = ' ';		
		strcpy(odR.ObjectName, queue_reply);

		fgets(reply_time, 48, fptr);			
		queuenamelen = strlen(reply_time) - 1;						
		reply_time[queuenamelen] = ' ';		
		fclose (fptr);
		}	   
   strcpy(QManager, "");	/*   Работаем с менеджером очередей по умолчанию  */
   MQCONN(QManager, Hcon, CompCode, CReason);              								  
   if (CompCode == MQCC_FAILED)
	{
     printf("MQCONN to %s ended with reason code %ld\n", QManager, CReason);
     exit(CReason);
	}
   
   O_options = MQOO_BROWSE + MQOO_INPUT_SHARED ;
   MQOPEN(Hcon, odG, O_options, Hobj, CompCode, Reason);             
   if (Reason != MQRC_NONE) { printf("MQOPEN %s ended with reason code %ld\n", queue_input, Reason); }
   if (CompCode == MQCC_FAILED) { exit(Reason);  }

   O_options = MQOO_BROWSE + MQOO_INPUT_SHARED ;
   MQOPEN(Hcon, odR, O_options, Hrep, CompCode, Reason);             
   if (Reason != MQRC_NONE) { printf("MQOPEN %s ended with reason code %ld\n", queue_reply, Reason); }
   if (CompCode == MQCC_FAILED) { exit(Reason);  }
   
   O_options = MQOO_OUTPUT ;
   MQOPEN(Hcon, odP, O_options, Hout, CompCode, Reason);             
   if (Reason != MQRC_NONE) { printf("MQOPEN %s ended with reason code %ld\n", queue_output, Reason); 	}
   if (CompCode == MQCC_FAILED) { exit(Reason);  }
    
   while (CompCode == MQCC_OK)
   {	 
    buflen = sizeof(buffer) - 1;    
     memcpy(md.MsgId, MQMI_NONE, sizeof(md.MsgId));
     memcpy(md.CorrelId, MQCI_NONE, sizeof(md.CorrelId));	 
	 gmo.Options = MQGMO_ACCEPT_TRUNCATED_MSG + MQGMO_WAIT + MQGMO_SYNCPOINT;   				 
	 gmo.WaitInterval = 3000 ;	 

	MQBEGIN (Hcon, mbo, CompCode, Reason);
    MQGET(Hcon, Hobj, md, gmo, buflen, buffer, messlen, CompCode, Reason);      	
	//if (Reason != MQRC_NONE) { printf("MQGET from %s ended with reason code %ld\n", queue_input, Reason);	}

	if ((CompCode == MQCC_OK)  ||  (CompCode == MQCC_WARNING))
			{				
			buffer[messlen] = '\0';  /* заносим символ конец строки в буфер с прочитанным сообщением */       
			buflen = messlen;	
				md.MsgType  = MQMT_REQUEST; 
			    md.Report = MQRO_EXCEPTION_WITH_DATA;
				strncpy(md.ReplyToQ, queue_reply, MQ_Q_NAME_LENGTH);
				memcpy(md.Format, MQFMT_STRING, MQ_FORMAT_LENGTH);

			MQPUT(Hcon, Hout, md, pmo, buflen, buffer,  CompCode, Reason);      
      if (Reason != MQRC_NONE) 
	{ 
	printf("MQPUT  to %s ended ended unsuccessfully with reason code %ld CompCode %ld\n", queue_output, Reason, CompCode );	
	MQBACK( Hcon, CompCode, Reason ) ;	
	CompCode = MQCC_FAILED ;
	}		
        else
       {

       while (CompCode != MQCC_FAILED)
       {
         /** осуществляется проверка queue_reply **/
		gmo.Options = MQGMO_ACCEPT_TRUNCATED_MSG + MQGMO_WAIT ;   				 
		gmo.WaitInterval = 3000 ;	 
		memcpy(md.MsgId, MQMI_NONE, sizeof(md.MsgId));
		memcpy(md.CorrelId, MQCI_NONE, sizeof(md.CorrelId));	 
	 MQGET(Hcon, Hrep, md, gmo, buflen, buffer, replylen, CompCode, Reason);      	
         if (CompCode != MQCC_FAILED)
         {       
			if (md.MsgType == MQMT_REPLY)     /* report feedback     */
			{	printf("Transaction % s=> %s successfully: %s\n", queue_input, queue_output, buffer);
				MQCMIT( Hcon, CompCode, Reason ) ;	
			}
			else
			{
				printf("Transaction % s=> %s successfully, REPLY message  not deliver, reason code %ld CompCode %ld\n", queue_input, queue_output, queue_reply, Reason, CompCode );				
				MQBACK( Hcon, CompCode, Reason ) ;	
				CompCode = MQCC_FAILED ;
			}
         }

		  if (Reason == MQRC_NO_MSG_AVAILABLE)
		  {
			  printf("Transaction % s=> %s UNsuccessfully, REPLY message  not deliver\n", queue_input, queue_output );	
			  MQBACK( Hcon, CompCode, Reason ) ;	
			  CompCode = MQCC_FAILED ;
		  }
               }
          }	
     }		
  }    
   C_options = 0;              
	MQCLOSE(Hcon, Hobj, C_options, CompCode, Reason);    
	if (Reason != MQRC_NONE){printf("MQCLOSE %s ended with reason code %ld\n", queue_input, Reason); }
	MQCLOSE(Hcon, Hout, C_options, CompCode, Reason);    
	if (Reason != MQRC_NONE){printf("MQCLOSE %s ended with reason code %ld\n", queue_output, Reason); }	
	MQCLOSE(Hcon, Hrep, C_options, CompCode, Reason);    
	if (Reason != MQRC_NONE){printf("MQCLOSE %s ended with reason code %ld\n", queue_reply, Reason); }	

   MQDISC(Hcon, CompCode, Reason);     
   if (Reason != MQRC_NONE){ printf("MQDISC ended with reason code %ld\n", Reason); }
   return(0);
 }

По тексту программы следует дать комментарии. Наличие опции

gmo.Options = MQGMO_SYNCPOINT;

подразумевает, что команда MQBEGIN может не указываться. Операторы

md.MsgType  = MQMT_REQUEST; 
strncpy(md.ReplyToQ,
        queue_reply,
        MQ_Q_NAME_LENGTH);

определяют тип сообщения REQUEST и очередь ответа, заданную в QUEUE_REPLY.

На очередь QUEUE_OUTPUT (или на удаленную очередь на другом менеджере) должна быть навешена программа-триггер, который возвращает сообщения типа Reply . Если Reply-сообщение поступает в очередь QUEUE_REPLY, то транзакция завершается успешно, в противном случае производится откат транзакции и сообщение восстанавливается в очереди QUEUE_INPUT. Reply-сообщение должно иметь идентификатор CorrelId такой же, как и MsgId исходного сообщения. В данной версии программы в целях упрощения отладки не проверяется это условие и читателю предлагается самостоятельно дописать этот фрагмент кода после отладки текущей версии программы. Работа с MsgId и CorrelId будет рассмотрена подробнее в лекции 11.

Программу-триггер, которая "навешивается" на очередь QUEUE_OUTPUT (или на удаленную очередь) для формирования Reply-сообщения ( md.MsgType = MQMT_REPLY; ), читателю также предлагается сделать самостоятельно.

На данном примере мы познакомились с WebSphere MQ транзакциями, являющимися основой создания надежных программ для передачи сообщений. Если сообщение приходит на сервер в очередь, то программа опроса очереди открывает внешнюю транзакцию для работы с WebSphere MQ и передает управление подпрограмме записи сообщения в базу данных, которая открывает внутреннюю транзакцию для работы с базой данных (БД). Если сообщение уходит из базы данных, то открывается внешняя транзакция работы с БД, далее открывается внутренняя транзакцию для работы с WebSphere MQ и идет помещение сообщения в очередь, из которой это сообщение "улетает" на другой сервер. Завершение транзакций и откат транзакций обоих типов осуществляется взаимосвязанно. Это и есть правильный стиль интеграции приложений на основе WebSphere MQ.

Списки распространения ( модель "один ко многим" )

Использование механизма списков распространения (Distribution List) или так называемой модели "один ко многим" требуется, например, в случае рассылки большому количеству клиентов постоянно меняющейся информации (котировки акций, курсы валют, новости и т.п.). Этот механизм позволяет одной командой MQOPEN открыть множество очередей и одной командой MQPUT положить сообщения в эти очереди. После открытия очередей возвращается один уникальный идентификатор объекта и MQPUT помещает сообщения во все эти очереди, используя этот единственный идентификатор .

В версии WebSphere MQ 5.1 и выше object descriptor (MQOD) содержит поля, которые используются для списков распространения. Поле Object Descriptor RecsPresent содержит число Object Records (MQORs) и если оно больше чем 0, то это означает, что должен быть использован список распространения.

Рассмотрим этот механизм на примере задачи, когда WebSphere MQ server помещает сообщения в N очередей, как показано на рис.9.3. Эти сообщения могут дальше уходить через remote queue или их может забирать WebSphere MQ client с заданной периодичностью. Назовем нашу программу distlist.exe, файл с текстом сообщения distlist.dat и файл инициализации distlist.ini, в котором 1-я строка – имя менеджера, 2-я и последующие строки – имена очередей, как показано ниже.

QM_ ALFA
Queue_ Moscow
Queue_ Kiev
Queue_ Alma-Ata	
Queue_ SPetersburg 
Queue_ Novosibirsk
Queue_ Saratov

//last string must be blank

Ниже приводится листинг программы distlist.cpp для Microsoft Visual C++ ver.6.0.

/* Листинг программы distlist                                       */
/* Program name: Distlist                                           */
/* Description: Distlist C program  pass messages to output queues  */
/* by Distribution list  for indicated Queue Manager                */
/* distlist.ini file give list of queue and distlist.dat give file  */
/* of message which copied to the output queue                      */

 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h> 
 #include <io.h>
 #include <time.h>
 #include <cmqc.h>
 char queue[1000][48] ; 
 char buf[48];  
 int  queuenamelen; 
 time_t tmr;
 FILE * fp; 
 FILE *fptr;	

static void print_usage(void);
static void print_responses( char * comment, PMQRR  pRR, MQLONG NumQueues, PMQOR  pOR);

 int main(int argc, char **argv)
 {
   typedef enum {False, True} Bool;
   MQOD     od = {MQOD_DEFAULT};    /* Object Descriptor             */
   MQMD     md = {MQMD_DEFAULT};    /* Message Descriptor            */
   MQPMO   pmo = {MQPMO_DEFAULT};   /* put message options           */
   MQHCONN  Hcon;                   /* connection handle             */
   MQHOBJ   Hobj;                   /* object handle                 */
   MQLONG   O_options;              /* MQOPEN options                */
   MQLONG   C_options;              /* MQCLOSE options               */
   MQLONG   CompCode;               /* completion code               */
   MQLONG   OpenCode;               /* MQOPEN completion code        */
   MQLONG   Reason;                 /* reason code                   */
   MQCHAR48 QManager;               /* queue manager name            */
   MQLONG   buflen;                 /* buffer length                 */
   char     buffer[101];            /* message buffer                */   
   MQLONG   Index ;                 /* Index into list of queues     */
   MQLONG   NumQueues ;             /* Number of queues              */ 
   PMQRR    pRR=NULL;               /* Pointer to response records   */
   PMQOR    pOR=NULL;               /* Pointer to object records     */
   Bool     DisconnectRequired=False;/* Already connected switch     */
   Bool     Connected=False;        /* Connect succeeded switch     */
      
   typedef struct 
   {
    MQBYTE24 MsgId;
    MQBYTE24 CorrelId;
   } PutMsgRec, *pPutMsgRec; 
   pPutMsgRec pPMR=NULL;            /* Pointer to put msg records    */    
   MQLONG PutMsgRecFields=MQPMRF_MSG_ID | MQPMRF_CORREL_ID;     

                           /* Open ini file and setting value  */
   	if ( (fptr=fopen ("distlist.ini","r" )) == NULL )
		{printf("Cannot open distlist.ini file" ); 
		 print_usage();	
		 exit(1);	}
	else{					
		fgets(QManager, 48, fptr);		
		queuenamelen = strlen(QManager) - 1;	
		QManager[queuenamelen] = ' ';
		NumQueues = 0;
	while (queuenamelen != 0)
		{	
		fgets(queue[NumQueues], 48, fptr);	
		queuenamelen = strlen(queue[NumQueues]) - 1;				
		queue[NumQueues][queuenamelen] = ' ';
		NumQueues++;
		}
	}
	fclose (fptr);	
	--NumQueues;		    /* NumQueues - Number of Queue name   */
   /*   Allocate response records, object records and put message records */   
   pRR  = (PMQRR)malloc( NumQueues * sizeof(MQRR));
   pOR  = (PMQOR)malloc( NumQueues * sizeof(MQOR));
   pPMR = (pPutMsgRec)malloc( NumQueues * sizeof(PutMsgRec));
   
   if((NULL == pRR) || (NULL == pOR) || (NULL == pPMR))
   {
     printf("%s(%d) malloc failed\n", __FILE__, __LINE__);
     exit(4);
   }

   /*   Use parameters as the name of the target queues              */
   
   for( Index = 0 ; Index < NumQueues ; Index ++)
   {
     strncpy( (pOR+Index)->ObjectName, queue[Index], (size_t)MQ_Q_NAME_LENGTH);		
     strncpy( (pOR+Index)->ObjectQMgrName, QManager, (size_t)MQ_Q_MGR_NAME_LENGTH);
   }   

   for( Index = 0 ; Index < NumQueues ; Index ++)
   {
     MQCONN((pOR+Index)->ObjectQMgrName, Hcon, ((pRR+Index)->CompCode), ((pRR+Index)->Reason)); 
     if ((pRR+Index)->CompCode == MQCC_FAILED)
     {
       continue;
     }
     if ((pRR+Index)->CompCode == MQCC_OK)
     {
       DisconnectRequired = True ;
     }
     Connected = True;
     break ;
   }
     /* Print any non zero responses                                   */
      print_responses("MQCONN", pRR, Index, pOR);

      /* Print If failed to connect to  queue manager then exit.        */   
   if( False == Connected  )
   {
     printf("Unable to connect to queue manager\n");
     exit(3) ;
   } 
   
   if ( (fp=fopen ("distlist.dat","r" )) == NULL )
		{printf("Cannot open distlist.dat file" ); exit(2);	}
	else{					
		fgets(buffer, 100, fptr);			
		buflen = (MQLONG)strlen(buffer);               /* length without null         */
       if (buffer[buflen-1] == '\n')		                  	/* last char is a new-line     */
       {
         buffer[buflen-1]  = '\0';				/* replace new-line with null  */
         --buflen;					/* reduce buffer length        */
       }
	}
		fclose (fp);	

   		tmr = time(NULL);
		strcpy ( buf, ctime(tmr));
		buf[strlen(buf)-5]=0;	
		printf("Distlist start  send message to list queue %s\n", buf);	

   /*   Open the target message queue for output                     */   
   od.Version = MQOD_VERSION_2 ;
   od.RecsPresent = NumQueues ;      
   od.ObjectRecPtr = pOR;            
   od.ResponseRecPtr = pRR ;
   O_options = MQOO_OUTPUT + MQOO_FAIL_IF_QUIESCING; 

   MQOPEN(Hcon, od,  O_options,  Hobj,  OpenCode,  Reason); 
   if (Reason == MQRC_MULTIPLE_REASONS)
   {
     print_responses("MQOPEN", pRR, NumQueues, pOR);
   }
   else
   {
     if (Reason != MQRC_NONE)
     {
       printf("MQOPEN returned CompCode=%d, Reason=%d\n", OpenCode, Reason);
     }
   }
      
   /*   Read message from the file 
   /*   Loop until null line or end of file, or there is a failure   */      
   CompCode = OpenCode;        /* use MQOPEN result for initial test */   
   		
   pmo.Version = MQPMO_VERSION_2 ;
   pmo.RecsPresent = NumQueues ;
   pmo.PutMsgRecPtr = pPMR ;
   pmo.PutMsgRecFields = PutMsgRecFields ;
   pmo.ResponseRecPtr = pRR ;  
   
           /*   Put  buffer to the message queue                       */
     if (buflen > 0)
     {
       for( Index = 0 ; Index < NumQueues ; Index ++)
       {
         memcpy( (pPMR+Index)->MsgId, MQMI_NONE, sizeof((pPMR+Index)->MsgId));
         memcpy( (pPMR+Index)->CorrelId, MQCI_NONE, sizeof((pPMR+Index)->CorrelId));
       }
       memcpy(md.Format, MQFMT_STRING, (size_t)MQ_FORMAT_LENGTH);

       MQPUT(Hcon, Hobj, md, pmo, buflen, buffer, CompCode, Reason); 
   
       if (Reason == MQRC_MULTIPLE_REASONS)
       {
         print_responses("MQPUT", pRR, NumQueues, pOR);
       }
       else
       {
         if (Reason != MQRC_NONE)
         {
           printf("MQPUT  returned CompCode=%d, Reason=%d\n", OpenCode, Reason);
         }
       }
		tmr = time(NULL);
		strcpy ( buf, ctime(tmr));
		buf[strlen(buf)-5]=0;		// strip new line
		printf("Distlist finish send message to list queue %s\n", buf);	

     }
     else   /* satisfy end condition when empty line is read */
       CompCode = MQCC_FAILED;
   //}
   
   if (OpenCode != MQCC_FAILED)
   {
     C_options = 0;
     MQCLOSE(Hcon, Hobj, C_options, CompCode, Reason); 
     if (Reason != MQRC_NONE)
     {
       printf("MQCLOSE ended with reason code %d\n", Reason);
     }
   }

   if (DisconnectRequired==True)
   {
     MQDISC(Hcon, CompCode, Reason);          
     if (Reason != MQRC_NONE)
     {
       printf("MQDISC ended with reason code %d\n", Reason);
     }
   }

   
   if( NULL != pOR )
   {
     free( pOR ) ;
   }
   if( NULL != pRR )
   {
     free( pRR ) ;
   }
   if( NULL != pPMR )
   {
     free( pPMR ) ;
   }
   return(0);
 }

static void print_usage(void)
{
 printf("Distlist correct usage is:\n\n");
 printf("Distlist Qmgr QName1 [QName2 [QName3 [...]]]\n\n");
}

static void print_responses( char * comment, PMQRR  pRR, MQLONG NumQueues, PMQOR  pOR)
{
  MQLONG Index;
  for( Index = 0 ; Index < NumQueues ; Index ++ )
  {
    if( MQCC_OK != (pRR+Index)->CompCode )
    {
      printf("%s for %.48s( %.48s) returned CompCode=%d, Reason=%d\n"
            , comment
            , (pOR+Index)->ObjectName
            , (pOR+Index)->ObjectQMgrName
            , (pRR+Index)->CompCode
            , (pRR+Index)->Reason);
    }
  } 
}

В завершение раздела можно сказать, что время работы механизма Distribution List для WebSphere MQ с 200, 400, 600 и т.д. очередями не зависит от производительности компьютера и не сильно зависит от количества очередей (для Notpersistent queue, persistent queue не целесообразно использовать для данной задачи ). Это наглядно видно из следующей таблицы, отражающей время работы distlist (сек) в зависимости от оперативной памяти (ОП) компьютера: 512Мбт и 1Гбт.

Таким образом, WebSphere MQ дает нам удобный механизм рассылки постоянно обновляющейся информации для 200, 400, 600 … клиентов. Верхняя граница числа очередей (клиентов) зависит от операционной системы и оперативной памяти компьютера. Число клиентов (очередей) ограничивается 1200 на Windows компьютере с памятью 512Мбт.

Задачи, решаемые с помощью механизмов списков распространения (Distribution List), могут быть успешно решены с помощью модели публикация/подписка (Publish/Subscribe), которая будет рассмотрена подробно в следующей лекции.