DBMS_SHEDULER. Запуск последовательных и зависимых заданий::Журнал СА 102009

АНТОН ПИЩУЛИН, главный специалист ОАО «ОТП Банк», занимается разработкой и интеграцией корпоративных информационных систем

DBMS_SHEDULER
Запуск последовательных и зависимых заданий

В корпоративных информсистемах бывает необходимо выполнить расчеты для отчетов и витрин данных для конечных пользователей.

Зачастую эти все вычисления нельзя запустить одним процессом, так как в ходе выполнения расчетов необходимо дождаться ряда событий (например, окончания выгрузки из одной системы в другую для какой-то определенной задачи). Можно, конечно, попытаться спрогнозировать, когда произойдут все необходимые события, и запустить весь расчет одним процессом, но в этом случае конечный пользователь увидит результат позже, что может повлиять на оперативность принимаемых решений особенно в условиях перерасчета большого объема данных.

В статье демонстрируется одно из возможных решений задачи оптимизации запуска связанных между собой процессов на платформе Oracle и объединенных в единую систему управления на основе пакета DBMS_SCHEDULER. В примере используется встроенный в Oracle механизм связанных задач (цепочки), объединенных вместе в одну пакетную задачу, которая запускается по сообщениям от клиента. Сообщения записываются и читаются клиентами из очереди Oracle. Все примеры кода приведены на PL/SQL, а все выполненные действия можно сделать как с помощью кода, так и с помощью Oracle EM (DBCONSOLE).

Немного теории

Для того чтобы приступить к демонстрации технологии связанных цепочек задач Oracle, приведем немного теории. В основе данной технологии лежит планировщик задач с расширенными функциями (расписаниями). Расписание в Oracle 10g является отдельной сущностью. Расписания бывают стандартными (старый вариант планировщика на основе пакета DBMS_JOB), на основе календаря (с возможностью задания расширенных правил на календарь), на основе событий (основан на чтении пользовательских сообщений из очереди). Последний вариант мы будем рассматривать далее, так как это соответствует нашей изначальной постановке задач.

Любая задача (Job) Oracle может быть привязана к расписанию. В свою очередь одна задача (Job) Oracle 10g может быть нескольких типов. Нас интересует прежде всего пакетный тип, когда в качестве запускаемого кода выбирается цепочка процессов (CHAIN). Процессы в цепочке могут быть простыми программами (процедурами) Oracle или расписаниями, или другими цепочками. Мы будем рассматривать самый простой – первый вариант. То есть предположим, что у нас есть три хранимые процедуры. Две первые из них должны запускаться параллельно при наступлении некоторого события, а третья дожидается выполнения их, только затем начинает свою работу. Для практического выполнения пользователю необходима привилегия dba.

Создание необходимых объектов очереди

Поскольку события запуска и остановки цепочки будут находиться в очереди сообщений Oracle, то на первом этапе нам необходимо создать необходимые для этого объекты (в Oracle EM Maintenance -> Management -> Messaging): тип сообщения и таблицы очереди, а также непосредственно самой очереди.

--Создание типа сообщения

create or replace type dwh_que_type as object

(

msg_id number,

msg varchar2(50)

);

--Создание таблицы DWH_QUE_TEST

begin

    dbms_aqadm.create_queue_table(

    queue_table => 'DWH_QUE_TEST',

    queue_payload_type => 'DWH.DWH_QUE_TYPE',

    sort_list => 'ENQ_TIME',

    multiple_consumers => true,

    message_grouping => sys.dbms_aqadm.transactional);

end;

--Создание очереди DWH_QUE_TST

begin

    dbms_aqadm.create_queue(

    queue_name => 'DWH_QUE_TST',

    queue_table => 'DWH_QUE_TEST',

    queue_type => sys.dbms_aqadm.normal_queue,

    max_retries => 500,

    retry_delay => 0,

    retention_time => 0);

end;

Создание расписания, цепочки и рабочих процедур

Далее переходим к созданию цепочки и расписания, а также макета рабочих (полезного бизнес кода) процедур, которые и будут участвовать в данном примере. А именно: будут созданы три процедуры, две из которых стартуют сразу, а третья дожидается выполнения первой и второй процедуры и начинает свою работу.

--Создание расписания TEST_SCH

--(в Oracle EM Administration > Schedules)

BEGIN

    dbms_scheduler.create_event_schedule(

    event_condition => 'tab.user_data.msg = ''TST_BEGIN''',

    queue_spec => '"DWH"."DWH_QUE_TST"',

    start_date => systimestamp at time zone 'Europe/Moscow',

    schedule_name => '"DWH"."TEST_SCH"');

END;

--Создание пустой цепочки DWH_CH_TST

--(в Oracle EM Administration > Chains)

BEGIN

sys.dbms_scheduler.create_chain(

chain_name => '"DWH"."DWH_CH_TST"');

END;

--Создания рабочих процедур

create or replace procedure p1

is

n number;

BEGIN

   n:=0;

   --Здесь нужно вставить рабочий код.

   --Если возникает ошибка, то ее обязательно нужно

   --обработать и сгенерить новый exception

exception when others then

   raise_application_error(-20112, 'p1');

END p1;

create or replace procedure p2

is

n number;

BEGIN

   n:=0;

   --Здесь нужно вставить рабочий код.

   --Если возникает ошибка, то ее обязательно нужно

   --обработать и сгенерить новый exception

exception when others then

   raise_application_error(-20112, 'p2');

END p2;

create or replace procedure p3

is

n number;

BEGIN

   n:=0;

   --Здесь нужно вставить рабочий код.

   --Если возникает ошибка, то ее обязательно нужно

   --обработать и сгенерить новый exception

exception when others then

   raise_application_error(-20112, 'p3');

END p3;

Создание программ Oracle

Теперь объединяем рабочие процедуры в программы, которые будут участвовать в качестве процессов в цепочке. Последняя программа pr3, кроме запуска рабочей процедуры p3, записывает сообщение (TST_END) окончания работы цепочки в очередь Oracle. Это сообщение будет прочитано клиентом, запустившим цепочку.

--Создание программ (в Oracle EM Administration > Programs)

BEGIN

DBMS_SCHEDULER.CREATE_PROGRAM(

program_name=>'"DWH"."pr1"',

program_action=>'begin

    p1();

    end;',

program_type=>'PLSQL_BLOCK',

number_of_arguments=>0,

enabled=>TRUE);

END;

BEGIN

DBMS_SCHEDULER.CREATE_PROGRAM(

program_name=>'"DWH"."pr2"',

program_action=>'begin

    p2();

    end;',

program_type=>'PLSQL_BLOCK',

number_of_arguments=>0,

enabled=>TRUE);

END;

--Последняя программа отличается от других

BEGIN

DBMS_SCHEDULER.CREATE_PROGRAM(

program_name=>'"DWH"."pr3"',

program_action=>'

    begin

      begin

      p3();

      end;

      --Далее идет сервисная часть, посылающая клиенту

      --сообщение о завершении работы цепочки

      DECLARE

        enqueue_options DBMS_AQ.enqueue_options_t;

        dequeue_options DBMS_AQ.dequeue_options_t;

        message_properties DBMS_AQ.message_properties_t;

        message_handle RAW(16);

        message DWH_QUE_TYPE;

      BEGIN

        message := DWH_QUE_TYPE(1, ''TST_END'');

        DBMS_AQ.ENQUEUE(

           queue_name => ''DWH_QUE_TST'',

           enqueue_options => enqueue_options,

           message_properties => message_properties,

           payload => message,

           msgid => message_handle);

        COMMIT;

      END;

    end;

    ',

program_type=>'PLSQL_BLOCK',

number_of_arguments=>0,

enabled=>TRUE);

END;

Создание шагов цепочки Oracle

Завершаем создание цепочки, то есть создаем шаги цепочки (S1, S2, S3) из созданных программ выше и определяем связи (правила) между ними. Чтобы цепочка начала свою работу, необходимо вначале (rule1) установить правило true для первых запускаемых программ. Второе правило (rule2) запускает шаг S3 при условии, что шаги S1 и S2 завершили свою работу. В конце создания цепочки необходимо ее активировать.

--Создание шагов цепочки

BEGIN

   dbms_scheduler.define_chain_step(

    chain_name => '"DWH"."DWH_CH_TST"',

    step_name => '"S1"',

    program_name => '"DWH"."pr1"');

END;

BEGIN

   dbms_scheduler.define_chain_step(

    chain_name => '"DWH"."DWH_CH_TST"',

    step_name => '"S2"',

    program_name => '"DWH"."pr2"');

END;

BEGIN

   dbms_scheduler.define_chain_step(

    chain_name => '"DWH"."DWH_CH_TST"',

    step_name => '"S3"',

    program_name => '"DWH"."pr3"');

END;

--Создание правил в цепочке

BEGIN

dbms_scheduler.define_chain_rule(

   chain_name => '"DWH"."DWH_CH_TST"',

   condition => 'true',

   rule_name => 'rule1',

   action => 'START "S1","S2"');

END;

BEGIN

dbms_scheduler.define_chain_rule(

   chain_name => '"DWH"."DWH_CH_TST"',

   condition => 'S1 COMPLETED AND S2 COMPLETED',

   rule_name => 'rule2',

   action => 'START "S3"');

END;

BEGIN

dbms_scheduler.define_chain_rule(

   chain_name => '"DWH"."DWH_CH_TST"',

   condition => 'S3 COMPLETED',

   rule_name => 'rule3',

   action => 'END');

END;

--Активация цепочки

BEGIN

dbms_scheduler.enable('"DWH"."DWH_CH_TST"');

END;

Создание пакетной задачи Oracle

Осталось только создать пакетную задачу определенного типа (chain), указать ей расписание (test_sch) и добавить пользователя dwh (под которым велась вся работа) в качестве подписчика очереди сообщений (dwh_que_tst). Затем остается только стартовать очередь и перейти к тестированию всей работы цепочки.

--Создание пакетной задачи

--(в Oracle EM Administration > Jobs)

BEGIN

dbms_scheduler.create_job(

   job_name => '"DWH"."JOB_TST"',

   job_type => 'CHAIN',

   job_action => '"DWH"."DWH_CH_TST"',

   schedule_name => '"DWH"."TEST_SCH"',

   job_class => '"DEFAULT_JOB_CLASS"',

   auto_drop => FALSE,

   enabled => FALSE);

   dbms_scheduler.set_attribute( name => '"DWH"."JOB_TST"',

    attribute => 'logging_level',

    value => DBMS_SCHEDULER.LOGGING_FULL);

   dbms_scheduler.enable( '"DWH"."JOB_TST"' );

END;

--Добавление пользователя dwh как подписчика очереди

--сообщений(в Oracle EM Maintenance > Management >

--Messaging)

declare

subscriber sys.aq$_agent;

begin

subscriber := sys.aq$_agent('"DWH"', '', 0);

dbms_aqadm.add_subscriber(

    queue_name => '"DWH"."DWH_QUE_TST"',

    subscriber=> subscriber, rule=> '' ,

    transformation=> '');

end;

--Запуск очереди

begin

DBMS_AQADM.START_QUEUE(

   'DWH_QUE_TST');

end;

Тестирование работы цепочки

После помещения TST_BEGIN в очередь сообщений срабатывает расписание TEST_SCH и запускает пакетную задачу JOB_TST. Затем клиент читает очередь и ждет сообщения TST_END и завершает свою работу.

--Запуск цепочки

DECLARE

       enqueue_options     DBMS_AQ.enqueue_options_t;

       dequeue_options     DBMS_AQ.dequeue_options_t;

       message_properties DBMS_AQ.message_properties_t;

       message_handle      RAW(16);

       message             DWH_QUE_TYPE;

    BEGIN

       message := DWH_QUE_TYPE(1, 'TST_BEGIN');

       --Помещение в очередь сообщения TST_BEGIN

       DBMS_AQ.ENQUEUE(

           queue_name => 'DWH.DWH_QUE_TST',

           enqueue_options => enqueue_options,

           message_properties => message_properties,

           payload => message,

           msgid => message_handle);

       COMMIT;

    /*Ожидание сообщения о завершении работы цепочки*/

     DECLARE

       enqueue_options     DBMS_AQ.enqueue_options_t;

       dequeue_options     DBMS_AQ.dequeue_options_t;

       message_properties DBMS_AQ.message_properties_t;

       message_handle      RAW(16);

       message             DWH_QUE_TYPE;

       s                   varchar(20);

       s2                  varchar2(20);

     BEGIN

       s2 := 'TST_END';

       s := ''''||s2||'''';

       dequeue_options.consumer_name := 'DWH';

       dequeue_options.wait := DBMS_AQ.FOREVER;

       dequeue_options.navigation := DBMS_AQ.FIRST_MESSAGE;

       dequeue_options.deq_condition := ?

           'tab.user_data.msg = '||s;

       --Читаем очередь, ожидаем сообщения TST_END

       DBMS_AQ.DEQUEUE(

           queue_name => 'DWH.DWH_QUE_TST',

           dequeue_options => dequeue_options,

           message_properties => message_properties,

           payload => message,

           msgid => message_handle);

       COMMIT;

    END;

END;

***

В статье была продемонстрирована возможность по оптимизации запуска связанных между собой процессов Oracle. В основе демонстрации был использован функционал Oracle 10g (пакет dbms_scheduler), позволяющий управлять цепочкой задач при помощи определенных правил и очередей сообщений Oracle. За более подробной информацией (описание пакета, атрибутов, свойств и т.д.) необходимо обращаться к официальной документации Oracle [1].

http://download.oracle.com/docs/cd/B19306_01/appdev.102/b14258/d_sched.htm.