Интеграция приложений на основе WebSphere MQ
Лекция #9: Основы программирования для WebSphere MQ
Лекция посвящена написанию программ, работающих с WebSphere MQ на основе интерфейса MQI. Первая программа rewriter (модель "один к одному") предназначена для чтения сообщений из одной очереди, записи их в другую очередь и в файл на диске. Рассматривается технология разработки приложений с использованием транзакций WebSphere MQ и транзакций базы данных. Приводится программа transmit, работающая с транзакционными функциями MQBEGIN, MQCMIT, MQBACK. Третья программа лекции distlist демонстрирует работу с механизмом списков распространения (Distribution List), реализующим модель "один ко многим". Этот механизм обычно используется в случае рассылки большому количеству клиентов постоянно меняющейся информации (котировки акций, курсы валют, новости и т.п.).

Программа rewriter (модель "один к одному")

Первая программа будет достаточно простая и реализует так называемую модель "один к одному" или "точка-точка". Эта программа предназначена для чтения сообщений из очереди 1, записи их в очередь 2 и лог-файл на диске. Эта программа имеет практическое значение. Достаточно часто необходимо иметь файл переданных сообщений за определенный период времени, чтобы быстро ответить на вопрос "Было ли передано сообщение с такими идентификационными параметрами в теле сообщения:…"? WebSphere MQ сохраняет persistent сообщения на диске, но эти лог-файлы малопонятны, предназначены для восстановления сообщений при сбоях и достаточно быстро перезаписываются менеджерами очередей при значение параметра logging = circular (по умолчанию) и больших потоках сообщений (logging = linear рекомендуется только для систем промышленной эксплуатации и в этом случае администратор WebSphere MQ должен заботиться о том, чтобы лог-файлы не "замусорили" весь жесткий диск). Поэтому наша программа может быть достаточно полезной.

Автору приходилось сталкиваться с "плохим" стилем программирования, когда параметры программы "зашиваются" в текст. Даже в учебных курсах этого следует избегать, несмотря на некоторое усложнение программ. В наших программах мы будем использовать простые файлы инициализации, чтобы избежать этой ошибки. Назовем нашу программу rewriter.exe и файл инициализации rewriter.ini, в котором 1-я строка – имя очереди для чтения, 2-я строка – имя очереди для записи, 3-я строка – имя лог-файла, как показано ниже.

QUEUE_INPUT
QUEUE_OUTPUT
C:\TEMP\rewriter.log

Разрабатываемая программа может быть представлена в следующей последовательности псевдокода:

    MQCONN
    MQOPEN	
--> цикл чтения сообщений
|   (на основе gmo.WaitInterval):  
|	 MQGET
|	 MQPUT
|-- конец цикла
    MQCLOSE
    MQDISC

Ниже приводится листинг программы rewriter.cpp для Microsoft Visual C++ ver.6.0. Не забудьте добавить mqm.Lib в Project => Settings => Link и обратиться к документации [15], [16], [17] в случае проблем с программированием.

/* Листинг программы rewriter */
/**********************************************************************/
 /* Program name: Rewriter                                            */
 /* Description: Rewriter C program  pass messages to output queue    */
 /* Function:                                                         */
 /* Rewriter is a sample C program to demonstrate the main MQI calls; */
 /* each message is copied from the input queue to the output         */
 /* queue, and sends a report to the log file                         */
/**********************************************************************/
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <time.h>
 #include <signal.h>
 #include <io.h>
     /*  includes for MQI  */
 #include <cmqc.h>
 char queue1[48] = "";
 char queue2[48] = "";
 char logfilename[48] = "";
 char logfilename2[48] = "";
 char buf[48];  
 int  queuenamelen;
 time_t tmr;
 FILE * fp; 
 FILE *fptr;	
 
 void cntrl_c_handler(int sig);

   /*   Declare MQI structures needed                                */
   MQOD    odG = {MQOD_DEFAULT};    /* Object Descriptor for GET     */
   MQOD    odP = {MQOD_DEFAULT};    /* Object Descriptor for PUT     */
   MQOD    odI = {MQOD_DEFAULT};    /* Object Descriptor for InitQ   */
   MQOD    odR = {MQOD_DEFAULT};    /* Object Descriptor for report  */
   MQMD     md = {MQMD_DEFAULT};    /* Message Descriptor            */
   MQGMO   gmo = {MQGMO_DEFAULT};   /* get message options           */
   MQPMO   pmo = {MQPMO_DEFAULT};   /* put message options           */     
   MQTMC2   *trig;                  /* trigger message structure     */
   MQCHAR48 QManager;               /* queue manager name            */
   MQHCONN  Hcon;                   /* connection handle             */
   MQHOBJ   Hobj;                   /* object handle, server queue   */
   MQHOBJ   Hinq;                   /* handle for MQINQ              */
   MQHOBJ   Hout;                   /* handle for MQPUT              */
   MQLONG   O_options;              /* MQOPEN options                */
   MQLONG   C_options;              /* MQCLOSE options               */
   MQLONG   CompCode;               /* completion code               */
   MQLONG   Reason;                 /* reason code                   */
   MQLONG   CReason;                /* reason code (MQCONN)          */   
   MQBYTE   buffer[8001];            /* message buffer               */
   MQLONG   buflen;                 /* buffer length                 */
   MQLONG   messlen;                /* message length received       */
   MQLONG   Select[1];              /* attribute selectors           */
   MQLONG   SelectValue[1];         /* value attribute selectors           */
   MQLONG  char_count;

   int main(int argc, char **argv)
 {     
    strcpy(QManager, "");	/*   Работаем с менеджером очередей по умолчанию  */
   if ( (fptr=fopen ("rewriter.ini","r" )) == NULL )
      {printf("Cannot open rewriter.ini file" ); exit(1);	}
   else{			/*   Открываем ini-файл и присваиваем значения переменным  */
      fgets(queue1, 48, fptr);
      queuenamelen = strlen(queue1) - 1;				
      queue1[queuenamelen] = ' ';
      fgets(queue2, 48, fptr);			
      queuenamelen = strlen(queue2) - 1;						
      queue2[queuenamelen] = ' ';
      fgets(logfilename, 48, fptr);
      queuenamelen = strlen(logfilename) - 1;				
      logfilename[queuenamelen] = ' ';
   		tmr = time(NULL);
      strcpy ( buf, ctime(&tmr));
      buf[strlen(buf)-1]=0;		// переход на новую строку
      strncat (logfilename, buf,10);	

      strcpy(odG.ObjectName, queue1);
      strcpy(odP.ObjectName, queue2);		
      fclose (fptr);	
      }	   	
   
   MQCONN(QManager, &Hcon, &CompCode, &CReason);     								   
   if (CompCode == MQCC_FAILED)
   {
     printf("MQCONN ended with reason code %ld\n", CReason);
     exit(CReason);
   } 
   O_options = MQOO_INPUT_SHARED + MQOO_FAIL_IF_QUIESCING; 
   MQOPEN(Hcon, &odG, O_options, &Hobj, &CompCode, &Reason);	/* открываем очередь для чтения - &odG   */
   if (Reason != MQRC_NONE) { printf("MQOPEN (input) ended with reason code %ld\n", Reason); }
   if (CompCode == MQCC_FAILED) { exit(Reason);  }


   O_options = MQOO_OUTPUT + MQOO_FAIL_IF_QUIESCING;    
    MQOPEN(Hcon, &odP, O_options, &Hout, &CompCode, &Reason);  	/* открываем очередь для записи - &odP   */	
   if (Reason != MQRC_NONE) { printf("MQOPEN (output) ended with reason code %ld\n", Reason);	}
   if (CompCode == MQCC_FAILED) { exit(Reason); }   

   
   fp = fopen (logfilename,"a");						
   if ( fp==NULL ){ printf("Cannot open log file %s\n", logfilename); }		
   printf("Rewriter(C) sending messages from %s to %s and to log-file %s \n",odG.ObjectName, odP.ObjectName, logfilename);	
         
   /*****************************************************************************/   
   /*   Читаем сообщения из QUEUE_INPUT и пишем в QUEUE_OUTPUT      */
   /*   до тех пор пока не встретим сообщение об ошибке                        */   
   /*****************************************************************************/
   buflen = sizeof(buffer) - 1;
   while (CompCode == MQCC_OK)
   {
     gmo.Options = MQGMO_ACCEPT_TRUNCATED_MSG  + MQGMO_WAIT;    
     gmo.WaitInterval = 3000;			/* Ожидаем новые сообщения 3 секунды           */         
     //gmo.WaitInterval = MQWI_UNLIMITED;  
     memcpy(md.MsgId, MQMI_NONE, sizeof(md.MsgId));
     memcpy(md.CorrelId, MQMI_NONE, sizeof(md.CorrelId));

   MQGET(Hcon, Hobj, &md, &gmo, buflen, buffer, &messlen, &CompCode, &Reason);   	
   if ((CompCode == MQCC_OK)  ||  (CompCode == MQCC_WARNING))
     {
      buffer[messlen] = '\0';  /* заносим символ конец строки в буфер с прочитанным сообщением */       
      buflen = messlen;	
        MQPUT(Hcon, Hout, &md, &pmo, buflen, buffer, &CompCode, &Reason); 
      if ((CompCode == MQCC_OK)  ||  (CompCode == MQCC_WARNING))
         {			
      		tmr = time(NULL);
         strcpy ( buf, ctime(&tmr));
         buf[strlen(buf)-1]=0;		// переход к новой строке			
         Reason = fprintf( fp, "%s: %s\n", buf, buffer );		
         }
     }				/* конец обработки входного сообщения         */
   }				/* конец цикла чтения/записи сообщений функциями   MQGET, MQPUT */	 

   fclose (fp);	
   C_options = 0;                   /* нет никаких опций при закрытии */
   MQCLOSE(Hcon, &Hobj, C_options, &CompCode, &Reason);     /* закрываем очередь для чтения   */
   if (Reason != MQRC_NONE)	
      {printf("MQCLOSE (input) ended with reason code %ld\n", Reason); }

   MQCLOSE(Hcon, &Hout, C_options, &CompCode, &Reason);    	/* закрываем очередь для записи   */
   if (Reason != MQRC_NONE)
      {printf("MQCLOSE (output) ended with reason code %ld\n", Reason); }
   
   if (CReason != MQRC_ALREADY_CONNECTED)
   {
     MQDISC(&Hcon,  &CompCode, &Reason);     									 
     if (Reason != MQRC_NONE){ printf("MQDISC ended with reason code %ld\n", Reason); }
   }

   return(0);
 }
 
Листинг 9.1. Rewriter C program pass messages to output queue

В данной версии мы выходим из цикла программы по опции gmo.WaitInterval = 3000, когда ожидаем сообщение в очереди в течении 3 сек, а его там нет (опция gmo.WaitInterval работает быстрее, чем если бы мы опрашивали очередь по собственному временному циклу). Другой вариант программы может быть таким. Задаем gmo.WaitInterval = MQWI_UNLIMITED; что соответствует gmo.WaitInterval= -1. Программа будет крутиться "бесконечно" до тех пор, пока мы не остановим её принудительно, например, нажатием клавиш CNTRL_C (стандартный останов). В этом случае нужно добавить обработчик прерываний по нажатию CNTRL_C потому, что при таком выходе объекты очереди останутся не закрытыми и идентификаторы объектов окажутся "зависшими" в виртуальной памяти компьютера. А это может привести к тому, что при повторном запуске программы эти "зависшие" идентификаторы будут мешать нормальному функционированию программы. Во втором варианте открытие и закрытие лог-файла необходимо также делать в обработчике прерываний или после каждой команды MQPUT, в противном случае лог-файл не будет формироваться. Следует отметить, что размер массива buffer ограничивает длину сообщения 8Кб и при появлении сообщений большей длины следует увеличить размер буфера.

Программа rewriter.exe работает достаточно быстро и сравнительные скорости работы данного алгоритма при длине сообщения 1Кб на компьютере INTEL Pentium 1.8Ггц приведены в таблице ниже.

Таблица 9.1.
Язык программы\тип очереди Not Persistent Persistent
С++ 1000 сооб/сек 400 сооб/сек
Visual Basic 6.0 200 сооб/сек 140 сооб/сек

Увеличение длины сообщения не ведет к пропорциональному уменьшению скорости. Эти исследования читатель может проделать самостоятельно. Реальные приложения, работающие с базами данных, имеют скорость обработки сообщений в 3-4 раза меньше.

Возвращаясь к вопросу о стилях программирования, следует отметить, что обработка кода ошибки является обязательным атрибутом качественного программирования и об этом не следует забывать. В нашей программе дается предупредительное сообщение и делается выход из программы. Если этого не сделать, то простая описка в rewriter.ini файле приведет к зависанию программы и мучительному поиску причин такого зависания, не говоря о других более сложных ситуациях, например, когда очередь открыта эксклюзивно другим приложением.

Для версии программы gmo.WaitInterval = MQWI_UNLIMITED полезно сделать вывод на экран передаваемых сообщений, чтобы наблюдать динамику работы созданного интерфейса. Таких улучшений может быть достаточно много и мы рассмотрим две достаточно полезные модификации.

  1. Программа rewriter может вызываться как MQSeries-триггер. Для этого параметры можно задать следующим образом. Входная очередь – это очередь, на которой определен триггеринг. Выходная очередь – это User Data в триггерном процессе и имя лог файла – это Environment Date в триггерном процессе. В этом случае код в начале программы будет такой.

    /* Код для вызова rewriter.exe 
       как MQSeries-триггер */
       int main(int argc,
                char **argv)
     {   if (argc > 1) 
         {trig = (MQTMC2*)argv[1];     
         strncpy(odG.ObjectName,
                 trig->QName,
                 MQ_Q_NAME_LENGTH);
         strncpy(queue1,
                 trig->QName,
                 MQ_Q_NAME_LENGTH);
         strncpy(QManager,
                 trig->QMgrName,
                 MQ_Q_MGR_NAME_LENGTH);     
         strncpy(odP.ObjectName,
                 trig->UserData,
           MQ_PROCESS_USER_DATA_LENGTH);
         strncpy(queue2, trig->UserData,
           MQ_PROCESS_USER_DATA_LENGTH); 
         strncpy(logfilename2,
                 trig->EnvData, 48);
     }
    

    Возможная модификация этого варианта - программа rewriter может вызываться с передачей параметров через командную строку и эту модификацию читатель может проделать самостоятельно.

  2. Программа rewriter может быть модифицирована в программу разветвитель mqsplitter.exe: чтение сообщений из очереди 1 и запись их в очередь 2, в очередь 3 и лог-файл на диске.

Можно сделать программу mqsplitter.exe на разных языках, например, на Visual Basic 6.0 с интерфейсом, показанным на рис.9.1, и сравнить производительность программ на разных языках, реализующих один и тот же алгоритм. Такая задача будет хорошим лабораторным практикумом.

Интерфейс программы mqsplitter на VB6

Рис. 9.1.  Интерфейс программы mqsplitter на VB6

Модификацию программы mqsplitter.exe читателю предлагается сделать самостоятельно и одновременно проверить идею создания "вечного двигателя". Для создания "вечного двигателя" понадобиться изменить исходный mqsplitter.ini файл следующим образом:

QUEUE_INPUT
QUEUE_OUTPUT1
QUEUE_OUTPUT2
C:\TEMP\ mqsplitter.log
->
QUEUE_INPUT
QUEUE_OUTPUT1
QUEUE_ INPUT
C:\TEMP\ mqsplitter.log

Если в очереди QUEUE_INPUT будет хотя бы одно сообщение, то ваша программа будет посылать сообщения в очередь QUEUE_OUTPUT1 до тех пор, пока не будет остановлена. Можно заложить в ini-файл программы пятый параметр: время опроса очереди, измеряемое в миллисекундах. Такая программа окажется весьма полезной при тестировании интерфейсов.

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

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

  • Необходимость иметь западную банковскую отчетность.
  • Возможность работы клиентов через Интернет (интернет-банкинг).
  • Гибкость и адаптивность к изменениям в области банковского законодательства.
  • Передовые программно-аппаратные решения и наилучшие показатели по критерию цена/(качество + производительность).

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

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

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

  1. Создать две программы обработчика, работающих на платформах АБС1 и АБС2 для отправки и приема сообщений через WebSphere MQ.
  2. Использовать WebSphere Business Integration Message Broker (сокращенно WebSphere BI Broker) или, иначе называемый, WebSphere MQ Integrator.
  3. Использовать заложенные в T24 средства интеграции с 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 могут быть четырех типов:

  • Datagram - простое сообщение, не требующее ответа;
  • Request - сообщение-запрос, которое ожидает сообщение-ответ (reply message);
  • Reply - сообщение-ответ на сообщение-запрос;
  • Report - сообщение, которое описывает такое событие, как появление ошибки.

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

Структура объектов WebSphere MQ
Рис. 9.2.  Структура объектов WebSphere MQ

Итак, последовательность псевдокода представляется следующим образом (обратите внимание на блок 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);
 }
Листинг 9.2. Программа transmit.cpp для Microsoft Visual C++ ver.6.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

Механизм Distribution List для  WebSphere MQ
Рис. 9.3.  Механизм Distribution List для WebSphere MQ

Ниже приводится листинг программы 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);
    }
  } 
}
Листинг 9.3. Программа distlist.cpp для Microsoft Visual C++ ver.6.0.

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

Количество очередей Время работы distlist (сек) при ОП 512Мбт Время работы distlist (сек) при ОП 1Гбт
200 1 1
400 1 1
600 1 1
800 2 1
1000 3 2
1200 3 2

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

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

 © Макушкин В.А., Володичев Д.С. Все права защищены.