Дополнительные особенности программирования для WebSphere MQ

Работа с MsgId и CorrelId

Теперь полезно изучить способы контроля доставки сообщений, тем более что для этого существуют специальные поля в дескрипторе сообщений: MsgId и CorrelId . Например, некоторое приложение отправляет запросы на обработку в очередь Queue1 и ожидает уведомления о доставке в очереди ReplyToQ = Queue2. Совершенно очевидно, что уведомления о доставке не обязаны появиться в очереди Queue2 в той же последовательности, в какой отправлялись запросы в очередь Queue1. Как же в таком случае отслеживать доставку сообщений?

Другой пример, приложение отправляет запросы в разные системы, в частности, по банковским счетам клиента и по проверке его кредитной истории и ожидает получить ответ от запрашиваемых систем в одну и ту же очередь для всех клиентов. Как же идентифицировать поступающие ответы? Самое простое - использовать идентификаторы MsgId и CorrelId , которые являются уникальными для каждого сообщения и создаются на основе времени выполнения команды MQPUT. Идентификаторы MsgId и CorrelId создаются менеджером, если они установлены в NULL, либо заданы опции MQPMO_NEW_MSG_ID и/или MQPMO_NEW_CORREL_ID ; в противном случае, MsgId и CorrelId создаются приложением.

Таким образом, задавая значения MsgId и CorrelId при выполнении MQPUT, появляется возможность автоматически получать значения MsgId и CorrelId при чтении командой MQGET. При этом не надо забывать, что в цикле считывания перед командой MQGET следует ставить обнуление этих переменных ( MsgId=MQMI_NONE и CorrelId=MQCI_NONE ), иначе можно получить ошибку считывания сообщений из очереди MQRC_NO_MSG_AVAILABLE, несмотря на то, что сообщения в очереди имеются.

Рассмотрим на примере, как работает MQGET с MsgId и CorrelId на практике. Пусть в некоторую очередь поступили сообщения в определенном порядке, как указано в таблице ниже.

Теперь варьируя MsgId и CorrelId , можно читать самые разные сообщения.

MQGET (MSGID= MQMI_NONE, CORRELID= MQCI_NONE) читает первое доступное сообщение независимо от MsgId и CorrelId , то есть сообщение 1.

MQGET (MSGID= MQMI_NONE, CORRELID= 3) может прочитать только одно сообщение, имеющее значение CorrelId = 3, это сообщение 3.

MQGET (MSGID= 2, CORRELID= MQCI_NONE) читает первое сообщение со значением MsgId = 2, это сообщение 4.

MQGET (MSGID= 2, CORRELID= 2) может прочитать только одно сообщение, имеющее уникальное сочетание MsgId = 2 и CorrelId = 2, это сообщение 5.

Таким образом, если никакие другие сообщения не поступят в очередь, то в очереди после четырех MQGET останутся 2 сообщения. Здесь уместен вопрос о производительности работы менеджера очередей при таком поиске сообщений. Если число сообщений в очереди не превышает 100, то время поиска будет незначительным, оно практически не зависит от числа сообщений. Но если число сообщений в очереди превысит 1000, то менеджер будет сканировать очередь и время поиска будет заметным. На OS/390 MQSeries администратор может определить MsgId или CorrelId (но не одновременно) как индекс и это существенно ускорит поиск нужного сообщения в очереди. Следует также отметить, что на других платформах (AS/400, HP_UX, AIX, Sun Solaris, Windows) в версии 5.3 WebSphere MQ можно использовать опции соответствия MatchOption: MQMO_MATCH_MSG_ID и MQMO_MATCH_CORREL_ID. Если эти опции соответствия не будут определены, то MsgId и CorrelId будут игнорироваться, как если бы использовались опции MQMI_NONE и MQCI_NONE и будет извлекаться очередное сообщение.

Сформулируем одно общее правило при разработке программ с контролем доставки сообщений с помощью MsgId и CorrelId : если приложение перемещает сообщение из входной очереди в выходную, то MsgId входного сообщения перемещается в CorrelId выходного сообщения и создается новый уникальный идентификатор MsgId выходного сообщения. На уровне псевдокода это правило можно записать следующим образом:

INPUT_MSG_DESC.MsgId = MQMI_NONE;
INPUT_MSG_DESC.CorrelId  = MQCI_NONE;
MQGET (..........);
    /* Обработка входного сообщения  */        
OUT_MSG_DESC.MsgId = MQMI_NONE;
OUT_MSG_DESC.CorrelId = INPUT_MSG_DESC.MsgId;
MQPUT  (..........);

Разрабатывая логику в цепи программ-обработчиков сообщений на разных платформах следует стремиться к тому, чтобы CorrelId содержал MsgId исходного запроса, а MsgId выходного сообщения содержал уникальный идентификатор, полученный наиболее важной в функциональном смысле выходной программой. На основе такой логики легко получить подтверждение на выходе, что исходный запрос был отработан, и промежуточные программы проставили свой MsgId в выходном сообщении. Это позволяет осуществить контроль времени прохождения запроса и получения сообщения-отчета.

В завершение раздела, уж если речь пошла о контроле доставки сообщений, необходимо отметить возможность использования полей context (контекст) в дескрипторе сообщения для контроля авторизации пользователя и обеспечения безопасности. Группа полей context состоит из 8 полей, их значения по умолчанию приводятся в таблице ниже.

Используя UserIdentifier и AccountingToken, а также опции для работы с полями для альтернативной авторизации, позволяющей одному пользователю посылать сообщения от имени другого, можно контролировать авторизацию пользователей, запускающих задачи. Эти права устанавливает MQSeries - администратор с помощью команды setmqaut. Об этом не следует забывать при разработке приложений.

Группировка и сегментация сообщений

До настоящего момента говорилось о физических сообщениях WebSphere MQ, и этого было вполне достаточно. Обычные сообщения длиной до 4Мб и сообщения максимальной длины в 100Мбт покрывают потребности любых приложений на 95%. Необходимость использования сверхбольших сообщений (> 100Мбт) или ограничения в приложениях на длину сообщений приводят к необходимости использовать понятия физических и логических сообщений, механизмы группирования и сегментации сообщений.

Физическое сообщение - наименьший блок информации, который может быть размещен в очереди. Одно физическое сообщение принято называть сегментом.

Логическое сообщение это либо отдельный блок информации приложения, либо один или несколько сегментов.

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

Пусть необходимо объединить в группу три сообщения. Для этого в дескрипторе сообщений полю MQMD_GroupId (24 байта) нужно присвоить уникальный идентификатор и записать его в каждом сообщении. Следующему полю MQMD_MsgSeqNumber присвоить значения 1, 2 и 3 для первого, второго и третьего сообщений, соответственно, как показано это на рис.11.1. Если сообщение не включено в группу, то для него GroupId будет NULL (MQGI_NONE), а в MsgSeqNumber будет значение 1. WebSphere MQ может автоматизировать присвоение MsgSeqNumber во время выполнения функции MQPUT, если в PMO задана опция MQPMO_LOGICAL_ORDER. В этом случае также понадобится установить значение в новое поле дескриптора сообщений: MQMD.MsgFlags, которое может иметь значения MQMF_MSG_IN_GROUP и MQMF_LAST_MSG_IN_GROUP. Если в GroupId было установлено значение MQGI_NONE, то для GroupId будет сгенерирован новый уникальный идентификатор при использовании MQPMO_LOGICAL_ORDER. И еще одно правило для вывода группы сообщений: свойства сообщений PERSISTENCE и NOT_PERSISTENCE не должны перемешиваться в одной группе.

Для пояснения изложенного следует привести фрагмент программы, использующей вывод сообщений в виде группы.

if ((CompCode == MQCC_OK) || 
    (CompCode == MQCC_WARNING))
  {
  pmo.Options = MQPMO_LOGICAL_ORDER + 
                MQPMO_SYNCPOINT ;
  memcpy(md.GroupId, "1", 
         sizeof(md.GroupId));
  memcpy(md. MsgFlags, "MQMF_MSG_IN_GROUP",
         sizeof(md. MsgFlags));

  //memcpy(md.MsgSeqNumber, 1, 
  //       sizeof(md.MsgSeqNumber));
  //memcpy(md.MsgFlags, MQMF_SEGMENT, 
  //       sizeof(md.MsgFlags));

/* блок формирования buffer                */
  MQPUT1(Hcon, odR, md, pmo, messlen, 
         buffer, CompCode, Reason);
  if (Reason != MQRC_NONE)   
    printf("MQPUT1 ended with reason 
            code %ld\n", Reason);
  }             
/* конец цикла чтения поступающих сообщений                */

memcpy(md. MsgFlags, 
       "MQMF_ LAST_MSG_IN_GROUP", 
       sizeof(md. MsgFlags));

/* блок формирования buffer                */
MQPUT1(Hcon, odR, md, pmo, messlen,
       buffer, CompCode, Reason);    
if (Reason != MQRC_NONE)   
  printf("MQPUT1 ended with 
          reason code %ld\n", Reason);

Процесс извлечения группы логических сообщений из очереди командой MQGET во многом будет зависеть от значений, заданных в опции GMO. Если задать значение MQGMO_ALL_MSGS_AVAILABLE, то сообщения будут доступны для чтения только, когда все сообщения группы поступят в очередь. Если задана опция MQGMO_LOGICAL_ORDER, то сообщения можно извлекать в порядке, определяемом MsgSeqNumber и только по одному сообщению из группы. Если опция MQGMO_LOGICAL_ORDER не задана, то приложение должно выбирать сообщения двумя способами:

Например, опция CORREL_ID означает, что должно выбираться сообщение, корреляционный идентификатор которого, соответствует значению поля CorrelId параметра MsgDesc функции MQGET. Это согласование является дополнением к любому другому соответствию, заданному в MatchOption. Если корреляционный идентификатор есть MQCI_NONE, то это равносильно тому, что опция MQMO_MATCH_CORREL_ID не определена.

Теперь следует дать расширенное определение сегмента сообщения. Сегментом сообщения называется физическое сообщение, идентифицируемое тройкой: GroupId, MsgSeqNumber и Offset. Из лекции 8 известно, что Offset это поле формата MQLONG в структурах MQOD и MQMD. Таким образом, одно логическое сообщение может состоять из нескольких сегментов, как показано на рис.11.1.

Разбиение на сегменты ( сегментация ) и обратная процедура, называемая реассемблированием (reassembly) может осуществляться менеджером очередей или приложением. Для начала следует рассмотреть, как это осуществляет менеджер очередей. Если дескриптор сообщения имеет в поле MsgFlag значение MQMF_SEGMENTATION_ALLOWED, и менеджер очередей определил, что сообщение является слишком большим для его MaxMsgLength или для значения параметра MaxMsgLength для очереди, то тогда менеджер осуществляет сегментацию. Никаких предположений о том, как выполняется разбиение данных, сделано быть не может.

Реассемблирование с помощью менеджера осуществляется, если при чтении сообщений командой MQGET была задана GMO-опция: MQGMO_COMPLETE_MESSAGE. В этом случае извлекаются все сегменты, и только полное логическое сообщение помещается в программный буфер. Следует быть внимательным при задании преобразования данных в sender-канале. Поскольку не все сегменты приходят одновременно, преобразование данных в канале в связи с разными кодовыми страницами может привести к ошибкам и рекомендуется задавать предобработку (UOW - unit of work) с помощью менеджера (опция MQRC_UOW_AVAILABLE ) или приложения (опция MQRC_UOW_NOT_AVAILABLE ).

Сегментация с помощью приложения возможна в двух случаях:

Для сегментации с помощью приложения необходимо включить для функции MQPUT опцию MQPMO_LOGICAL_ORDER и проследить за правильным включением в поле MsgFlags флажков MQMF_SEGMENT и MQMF_LAST_SEGMENT. Менеджер очередей позаботиться о назначении и поддержании GroupId, MsgSeqNumber и Offset.

Реассемблирование с помощью приложения осуществляется, если при чтении сообщений командой MQGET была задана GMO-опция: MQGMO_COMPLETE_MESSAGE и флажок MsgFlags имеет значение MQMF_SEGMENTATION_ALLOWED. В этом случае приложение извлекает сегменты самостоятельно в зависимости от ограничений на свой размер буфера. Опции MQGMO_ALL_MSGS_AVAILABLE и MQGMO_ALL_SEGMENTS_AVAILABLE позволят начать работу, если все сообщения или все сегменты поступят в очередь. Как только первое сообщение поступит в очередь, можно использовать MQGMO_LOGICAL_ORDER опцию, чтобы быть уверенным, что все последующие сегменты логического сообщения будут обработаны. Если опция MQGMO_LOGICAL_ORDER не задана, то приложение может выбирать сообщения на основе опций соответствия MatchOption, как это делалось для групп:

Приложение может определить окончание работы с сегментом с помощью возвращаемого MQGET значения поля MQGMO.SegmentStatus, которое может принимать следующие значения: MQSS_SEGMENT, MQSS_LAST_SEGMENT, MQSS_NOT_A_SEGMENT.

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

/*    пример использования функции MQPUT */
PMO.Options = MQGMO_LOGICAL_ORDER + 
              MQGMO_SYNCPOINT

/* помещаем в очередь первое логическое 
   сообщение*/
MQPUT (MQMF_MSG_IN GROUP + MQMF_SEGMENT)
MQPUT (MQMF_MSG_IN GROUP + MQMF_SEGMENT)
MQPUT (MQMF_MSG_IN GROUP + MQMF_LAST_SEGMENT)

/* помещаем в очередь второе  логическое 
   сообщение */
MQPUT (MQMF_MSG_IN GROUP + MQMFSEGMENT)

/* помещаем в очередь третье логическое 
   сообщение */
MQPUT (MQMF_LAST_MSG_IN GROUP + MQMF_SEGMENT)
MQPUT (MQMF_ LAST_MSG_IN GROUP + 
       MQMF_LAST_SEGMENT)
MQCMIT

/* пример использования функции MQGET */
GMO.Options = MQGMO_LOGICAL_ORDER + 
              MQGMO_SYNCPOINT + 
              MQGMO_ALL_MSGS_AVAILABLE + 
              MQGMO_WAIT
Do while (Not MQGS_LAST_MSG_IN GROUP and 
          Not MQSS_LAST_SEGMENT)
  MQGET (........)
/* обработка сегмента или законченного 
   логического сообщения  */
End
MQCMIT

Channel-exit программы

Channel - exits программы выполняются канальными агентами. Эти программы первыми и последними встречают и провожают сообщения. Именно поэтому их чаще всего используют для шифрования, проверки идентификации и аутентификации сообщений в целях безопасности.

Иногда возникает потребность применять Channel- exits программы, например, для того чтобы проследить прохождение сообщений и выполнить их логирование, выявить возникшие проблемы. Допустим, что стандартное ПО этого не делает, "вклиниваться" в него не представляется возможным, так как все входящие сообщения считывает из очереди программа-обработчик. В этом случае можно написать простую Channel- exits программу в виде библиотечной функции для UNIX или как DLL (Dynamic Linl Library) для Windows. Программе следует присвоить имя, к примеру, mqexit.dll с точкой входа MsgExit. Для работы программы, ее надо прописать в том канале, через который идут сообщения:

С помощью программы RUNMQSC это может быть сделано командой:

alter chl(mychannel) chltype(mychannel_type)
    msgexit(' mqexit(MsgExit)') 
    msgdata('\mydir\mqexit.ini')

Программа mqexit.dll и настроечный файл mqexit.ini помещаются в стандартной директории (mydir ) для Channel exits программ (для Windows это C:\Program Files\IBM\WebSphere MQ\Exits\, для UNIX HP_UX это /var/mqm/exits/, аналогичные пути существуют для других платформ).

Настроечный файл mqexit.ini может содержать имя лог файла, имя буферного файла для записи входящего сообщения; имя очереди, в которую перекладывается содержимое буферного файла; имя менеджера, ведь менеджеров может быть несколько на одном сервере.

В учебных целях стоит упростить задачу, не использовать mqexit.ini-файл и записать имя лог-файла непосредственно в Message Exit Data (например, C:\TEMP\mqexit.log ), а с сообщениями пусть разбирается прежняя программа-обработчик, читающая входящие сообщения из очереди. При этом предполагается, что используется менеджер очередей по умолчанию. В этом случае программа mqexit.dll для Windows выглядит следующим образом.

/* Листинг программы mqexit.dll */
/************************************************************************/
/* Program name: mqexit                                                                                                    */
/* Description: This is a sample message exit function record all messages             */
/* This function will work on any channels, e.g. send, receive, requester, etc.        */
/* This function logs all channel activities into the Channel Log file.                */
/* Please see Intercommunication documentation  to specify the Channel Message Exit    */
/* Authors: Vladimir Makushkin, Moscow, Russia                                         */
/*                 email:  vmakushkin@mail.ru                                          */
/*  Disclaimer: this program has been tested under HP_UX and Windows  to the           */
/*  authors' satisfaction.  You may use this program at your own risk.  Author are not */
/*  responsible  for any unexpected results generated by this program.                 */
/************************************************************************/
/* standard headers */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <locale.h>
#include <sys/types.h>
#include <sys/timeb.h>
#include <time.h>
/* MQSeries headers */
#include <cmqc.h>
#include <cmqxc.h>
char timebuf??(130??);
char Channel_Logfile??(129??);
char MsgBufFile??(129??);
char ini_filename??(129??);
FILE * OST = 0; 
#define DBGPRINTF(x) { \
               OST=fopen(Channel_Logfile,"a+"); \
               if(OST) fprintf x; \
               fclose(OST); \
               }
void writer (long *, char *); /* prototype for file writer function */
void getINI( void );
void MQENTRY MQStart(void) {;}
void MQENTRY MsgExit( PMQCXP     pChannelExitParams,
                  PMQCD      pChannelDefinition,
                  PMQLONG    pDataLength,
                  PMQLONG    pAgentBufferLength,
                  PMQBYTE    AgentBuffer,
                  PMQLONG    pExitBufferLength,
                  PMQPTR     pExitBufferAddr)
   {
    short i;
    struct timeb t;
    char millisec??(25??);    
    time_t clock = time( (time_t*) NULL);
    struct tm *tmptr = localtime(clock);
    strcpy(timebuf, asctime(tmptr));
    ftime( t );
    memset( millisec, 0, sizeof( millisec ) );
    sprintf( millisec, " Time:%ld.%d", t.time, t.millitm );
    strcat( timebuf, millisec );
    if ( strlen(pChannelExitParams->ExitData) )
    {
        memset( ini_filename, 0, sizeof( ini_filename ) );
        strncpy( ini_filename, pChannelExitParams->ExitData, MQ_EXIT_DATA_LENGTH );
    
        /*------------------------*/
        /* remove space from data */
        /*------------------------*/
        for ( i = MQ_EXIT_DATA_LENGTH - 1; i > 0; i-- ) 
            {
            if ( ini_filename??(i??) != ' '  ini_filename??(i??) )
                break;
            ini_filename??(i??) = 0;
            }
    } 
    getINI();    
void writer(long *lngth, char *bf)
{

long msgsize;
int n;
FILE *fp;
getINI();
fp = fopen(MsgBufFile, "ab");
msgsize = *lngth;
n = fwrite(msgsize, sizeof(msgsize), 1, fp); /* put record length */
n = fwrite(bf, msgsize, 1, fp);          /* and message buffer */
fclose(fp);
}

/*---------------------------------------------------------
   getINI() - read information from INI file
-----------------------------------------------------------*/
void getINI( void )
{
   FILE *ini_fp;
   char *ptr, *name_ptr;
   char field_val??(129??), Ini_s??(129??), fn??(129??);
   short len;
  
   memset( Ini_s, 0, sizeof( Ini_s ) );
   memset( field_val, 0, sizeof( field_val ) );
   memset( MsgBufFile, 0, sizeof ( MsgBufFile ) );
   memset( Channel_Logfile, 0, sizeof( Channel_Logfile ) );
    
   if ( strlen( ini_filename ) )                /* if Channel Exit defined INI filename */
    strcpy( fn, ini_filename );             /* use it */
   else                                         /* otherwise use default INI filename */
    strcpy( fn, INI_FILENAME );  
   if ( (ini_fp = fopen( fn, "r" )) == NULL )
      {
      fprintf(stderr, "\nUnable to open %s.  Error #: %d", fn, errno );
      return;
      }
    
   /*-----------------------------*/
   /* read the INI file until eof */
   /*-----------------------------*/
   while ( (ptr = fgets( Ini_s, sizeof(Ini_s) - 1, ini_fp )) != NULL )
   {
     len = strlen( Ini_s );
     if ( Ini_s??(len -1??) == LINEFEED )
    Ini_s??(len-1??) = 0;                               /* null out '\n' */
     if ( (name_ptr = strchr( Ini_s, EQUAL )) == NULL )     /* no '=' found */
     continue;
     ptr = name_ptr + 1;
     *name_ptr = 0;                                         /* set null for s[] */
     strcpy( field_val, ptr );
   fclose( ini_fp );

   return;
}

В качестве комментария к программе надо отметить, что обязательным является соблюдение трех рекомендаций для Windows из документа - Intercommunication.

#include <cmqc.h>
#include <cmqxc.h>
void MQStart() {;} 
    /* dummy entry point - for 
       consistency only */
void MQENTRY ChannelExit ( 
  PMQCXP pChannelExitParms,
  PMQCD pChannelDefinition,
  PMQLONG pDataLength,
  PMQLONG pAgentBufferLength,
  PMQVOID pAgentBuffer,
  PMQLONG pExitBufferLength,
  PMQPTR pExitBufferAddr)
{
... Insert code here
}

LIBRARY mqexit 
 DESCRIPTION 'Provides Retry 
              and Channel exits'
 HEAPSIZE   4096
 STACKSIZE  8192 
 EXPORTS  ChannelExit

И после этого проходящие по каналу сообщения начинают записываться в файл, определенный в Message Exit Data, включая заголовок и данные сообщения. Скорость прохождения сообщений с использованием Channel exits программ уменьшится за счет команд записи на диск. Читателю предлагается исследовать самостоятельно во сколько раз изменится скорость работы программы и найти пути её повышения.

В заключение лекции следует отметить, что остались механизмы, которые не нашли отражение в этой книге и могли бы представлять интерес для читателя-программиста, а именно:

С этими вопросами читателю предлагается ознакомиться самостоятельно по документации.

Ключевым моментом интеграции приложений является использование WebSphere Business Integration Message Broker $$\text{\textregistered}$$, сокращенно WBI Message Broker или просто WBI Broker (ранее - IBM MQSeries Integrator $$\text{\textregistered}$$) для управления и преобразования потоков сообщений. Этот продукт компании IBM будет рассмотрен далее.