Интеграция приложений на основе 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 сооб/сек
1. Увеличение длины сообщения не ведет к пропорциональному уменьшению скорости. Эти исследования читатель может проделать самостоятельно. Реальные приложения, работающие с базами данных, имеют скорость обработки сообщений в 3-4 раза меньше. Возвращаясь к вопросу о стилях программирования, следует отметить, что обработка кода ошибки является обязательным атрибутом качественного программирования и об этом не следует забывать. В нашей программе дается предупредительное сообщение и делается выход из программы. Если этого не сделать, то простая описка в rewriter.ini файле приведет к зависанию программы и мучительному поиску причин такого зависания, не говоря о других более сложных ситуациях, например, когда очередь открыта эксклюзивно другим приложением. Для версии программы gmo.WaitInterval = MQWI_UNLIMITED полезно сделать вывод на экран передаваемых сообщений, чтобы наблюдать динамику работы созданного интерфейса. Таких улучшений может быть достаточно много и мы рассмотрим две достаточно полезные модификации. Программа 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 может вызываться с передачей параметров через командную строку и эту модификацию читатель может проделать самостоятельно. Программа rewriter может быть модифицирована в программу разветвитель mqsplitter.exe: чтение сообщений из очереди 1 и запись их в очередь 2, в очередь 3 и лог-файл на диске. Можно сделать программу mqsplitter.exe на разных языках, например, на Visual Basic 6.0 с интерфейсом, показанным на рис.9.1, и сравнить производительность программ на разных языках, реализующих один и тот же алгоритм. Такая задача будет хорошим лабораторным практикумом. Рис. 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 и АБС2 для отправки и приема сообщений через WebSphere MQ. Использовать WebSphere Business Integration Message Broker (сокращенно WebSphere BI Broker) или, иначе называемый, WebSphere MQ Integrator. Использовать заложенные в 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 сек. Эта задача может использоваться в практических целях при нестабильной работе каналов, например выделенных. Наше приложение при откате транзакции может попытаться перенаправить сообщений из входной очереди – но это уже другая задача. Рис. 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
Рис. 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), которая будет рассмотрена подробно в следующей лекции.

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