【问题标题】:pl/sql using trigger to gather queued messages in queuepl/sql 使用触发器收集队列中的排队消息
【发布时间】:2014-07-15 08:58:34
【问题描述】:

我正在努力改变一个已经存在的排队系统......这让我很头疼......

队列背后的逻辑都在Oracle DB中,如下:

  1. 数据库 #1 发送通知,这些通知在数据库 #2 中的特定队列中排队,我们将其命名为 Q1。
  2. 有一个常规作业,它执行包中的一个过程 (DISPATCHER),它基本上是一个循环,它将 Q1 中的所有消息出列,直到没有消息存在。它每 5 分钟执行一次,但由于队列条目的数量,它运行的时间比这长。
  3. Dispatcher 过程将消息分发到其他队列,Java 从这些队列中处理消息,然后事情会更进一步。

对调度 JOB 没有太多控制,所以我决定寻找使整个框架更好地调度作业的方法。我必须补充一点,它是一个使用了很长一段时间的旧实现,但大多数人觉得它不容易控制或透明。

我的意思是这个调度过程是否有任何安全的方法?

  1. 我已经在队列表上创建了一个触发器,该触发器执行单个出队,所有消息处理都在 DISPATCHER 中完成,它可以工作,但我不知道这是否是最好的主意。
  2. 查找创建订阅者,但无法添加任何订阅者,因为它是单一消费者队列。考虑到我必须更新框架,重新创建队列会删除队列中已经存在的所有消息,这是我想省略的内容...
  3. 尝试了一个 DBMS 调度程序作业来执行调度程序以获得更多控制,但它只是在“有资源时”运行。我指望一种能够持续工作并尽可能快地处理消息的解决方案。
  4. 也许有一些更精细的方法来解决这个问题,但我还是有点新鲜,似乎无法掌握整个先进的排队机制......

提前致谢。

【问题讨论】:

  • 你为什么不使用像 Oracle DBMS_AQ 或 RabbitMQ 或 Redis 之类的消息队列?使用市场上已有的东西通常要容易得多。对调度工作没有任何控制权是好的;它们应该是松耦合的。如果你不能改变任何东西,那么使用 DBMS_SCHEDULER 可能是要走的路。考虑为其获取更多资源并增加允许运行的作业数量。
  • 这个框架大约在 8 到 10 年前就已经开发出来了……因为它是一个“到目前为止运行良好”的系统,没有人敢以任何方式对其进行修改。此外,更改此设置之后需要进行大量测试,这意味着需要大量时间,因此我正在寻找一个好的但不太复杂的解决方案。
  • 那么,是什么阻止您增加可以运行的作业数量并确保您有一个不断运行的作业?
  • DBMS 调度程序工作似乎运行良好,我不介意 DISPATCHER 不断工作,但有时需要关闭它,即在执行对数据库的部署时。由于某种原因,很难停止常规作业并按照设定的时间表重新开始。我正在考虑 DBMS 调度程序作业,但想检查所有可能的选项。

标签: oracle triggers queue


【解决方案1】:

考虑到不断运行的作业,可以通过DBMS_PIPE 发送消息来控制过程执行。
控制代码通过 dbms_pipe 发送stop 消息。
调度程序代码在DBMS_SCHEDULER 作业中不断运行,并检查stop 消息是否到达管道。
也可以在超时后强制中断 DBMS_SHCEDULER 作业。

几年前我已经实现了这样一个逻辑,这个解决方案一直运行到现在。下面的代码是一个经过修改的示例,可以帮助您检查这种方法是否可以接受并且对您的案例有用。

控制包头:

create or replace package DispatcherControl is
  -- Package to control job state      

  procedure StartDispatcher;

  procedure StopDispatcher;

end;

控制包体:

create or replace package body DispatcherControl is

  -- User who owns dispatcher job in DBMS_SCHEDULER
  C_DISPATCH_PROGRAM_OWNER    constant varchar2(100) := 'DISPATCH_USER';
  -- Main procedure to handle dispatcher requests
  C_DISPATCH_JOB_PROC         constant varchar2(100) := 'DISPATCH_USER.DISPATCHER.MAINPROC';
  -- Name for dispatch program
  C_DISPATCH_PROGRAM_NAME     constant varchar2(100) := 'DISPATCH_PROGRAM';
  -- Description for dispatch program
  C_DISPATCH_PROGRAM_COMMENT  constant varchar2(100) := 'Q1 dispatch task';
  -- Dispatcher job name.
  C_DISPATCH_JOB_NAME         constant varchar2(100) := 'DISPATCH_JOB';
  -- Description for dispatcher job
  C_DISPATCH_JOB_COMMENT          constant varchar2(100) := 'Q1 dispatch process';
  -- Pipe name for dispatch job control.
  C_DISPATCH_PIPE_NAME        constant varchar2(100) := 'DISPATCH_CONTROL';
  -- Message text for pipe to stop dispatch job
  C_PIPE_STOP_MSG constant varchar2(100) := 'stop_dispatch';

  -- Check if DBMS_SCHEDULER program exists and create it if needed
  procedure CheckDispatcherJobExists
  is
  begin

    -- Return if program exists
    for cDummy in (
      select 1
      from all_scheduler_programs
      where
        owner = C_DISPATCH_PROGRAM_OWNER and program_name = C_DISPATCH_PROGRAM_NAME
        and
        rownum = 1
    ) loop
      return;
    end loop;

    -- Create disabled if not found
    sys.dbms_scheduler.create_program(
      program_name        => C_DISPATCH_PROGRAM_OWNER || '.' || C_DISPATCH_PROGRAM_NAME,
      program_type        => 'STORED_PROCEDURE',
      program_action      => C_DISPATCH_JOB_PROC,
      number_of_arguments => 0,
      enabled             => false,
      comments            => C_DISPATCH_PROGRAM_COMMENT
    );

    -- Enable program
    sys.dbms_scheduler.enable(C_DISPATCH_PROGRAM_OWNER || '.' || C_DISPATCH_PROGRAM_NAME);

  end;


  -- Check status of dispatcher job and run it if not found
  procedure CheckDispatcherJobState
  is
  begin

    CheckDIspatcherJobExists;

    -- Check if job is currently running according to scheduler info
    for cDummy in (
      select * from all_scheduler_jobs
      where
        owner = C_DISPATCH_PROGRAM_OWNER
    and
    job_name = C_DISPATCH_JOB_NAME
    ) loop

      -- Job found, check if running
      for cDummy2 in (
        select * from all_scheduler_running_jobs
    where
    owner = C_CIS_PROGRAM_OWNER  and job_name = C_DISPATCH_JOB_NAME
      ) loop

         -- Check if job really running by checking sessions list
        for cDummy3 in (
          select 1
          from
            all_scheduler_running_jobs jobs,
            sys.v_$session             sessions
          where
            jobs.owner = C_DISPATCH_PROGRAM_OWNER  and jobs.job_name = C_DISPATCH_JOB_NAME
            and
            sessions.sid = jobs.session_id and sessions.process = jobs.slave_os_process_id
            and
            sessions.action = jobs.job_name and sessions.username = jobs.owner
            and
            sessions.module = 'DBMS_SCHEDULER'
        ) loop

           -- Ok, return
           return;

        end loop;

        -- No process found for running job, stop and delete task before recreation
        sys.dbms_scheduler.stop_job(C_DISPATCH_JOB_NAME, true);

      end loop;

      -- process found, but restart needed
      sys.dbms_scheduler.drop_job(C_DISPATCH_JOB_NAME, true);

    end loop;

    -- Create one-time running job with manual start
    sys.dbms_scheduler.create_job(
      job_name        => C_DISPATCH_JOB_NAME,
      program_name    => C_DISPATCH_PROGRAM_OWNER || '.' || C_DISPATCH_PROGRAM_NAME,
      enabled         => true,
      auto_drop       => true,
      start_date      => add_months(sysdate,1000),
      repeat_interval => null,
      end_date        => null,
      comments        => C_DISPATCH_JOB_COMMENT
    );

    -- Run created task
    sys.dbms_scheduler.run_job(
      job_name            => C_DISPATCH_JOB_NAME,
      use_current_session => false
    );

  end;


  -- Stop and drop dispatch job.
  procedure DropDispatchJob
  is
    vSendRC integer;
  begin

    -- Send request through DBMS_PIPE and wait for timeout.
    sys.dbms_pipe.reset_buffer;
    sys.dbms_pipe.pack_message(C_PIPE_STOP_MSG);
    vSendRC := sys.dbms_pipe.send_message(C_DISPATCH_PIPE_NAME,1);
    if(vSendRC = 0) then
      -- wait if sent Ok
      sys.dbms_lock.sleep(0.25);
    end if;

    -- force job stop (not done in case of drop_job)
    for cDummy in (
      select 1 from all_scheduler_running_jobs
      where
        owner = C_DISPATCH_PROGRAM_OWNER
        and
        job_name = C_DISPATCH_JOB_NAME
    ) loop

      begin
        sys.dbms_scheduler.stop_job(C_DISPATCH_JOB_NAME, true);
      exception
        when others then begin
          -- If sqlcode = -27366 then task finished, in other case it's unexpected
          if(SQLCODE != -27366) then
            raise;
          end if;
        end;
      end;

    end loop;

    -- delete job if exists, allow stopping
    for cDummy in (
      select 1 from all_scheduler_jobs
      where
        owner = C_DISPATCH_PROGRAM_OWNER
        and
        job_name = C_DISPATCH_JOB_NAME
    ) loop

      sys.dbms_scheduler.drop_job(C_DISPATCH_JOB_NAME, true);

    end loop;

    -- drop program if exists
    for cDummy in (
      select 1 from all_scheduler_programs
        where
          owner = C_DISPATCH_PROGRAM_OWNER
          and
          program_name = C_DISPATCH_PROGRAM_NAME
    ) loop

      sys.dbms_scheduler.drop_program(C_DISPATCH_PROGRAM_NAME, true);

    end loop;

    -- Clear DBMS_PIPE messages after job stop (cover case of forced stop).
    sys.dbms_pipe.purge(C_DISPATCH_PIPE_NAME);

  end;

  -- Start dispatcher process
  procedure StartDispatcher
  is
  begin

    -- Clear message queue before starting
    sys.dbms_pipe.purge(C_DISPATCH_PIPE_NAME);

    CheckDispatchJobState;

  end;

  -- Stop dispatcher process
  procedure StopDispatcher
  is
  begin
    DropDispatchJob;
  end;

end;

调度程序包头:

create or replace package DISPATCH_USER.Dispatcher is

  -- Main queue check procedure to run from job.
  procedure MainProc;

end;

调度程序包体:

create or replace package body DISPATCH_USER.Dispatcher is

  -- Normal wait time in seconds
  NORMAL_CHECK_INTERVAL constant number := 0.01;

  -- Checks if stop message received.
  function CheckIsStopped return boolean
  is
   vPipeRC integer;
   vMsg    varchar2(1024);
  begin
    -- check if message exists
    vPipeRC := sys.dbms_pipe.receive_message('DISPATCH_CONTROL',0);
    if(vPipeRC = 0) then
      -- check type of message content, must be varchar2 (look for constants in sys.dbms_pipe package).
      if(sys.dbms_pipe.next_item_type = 9) then

        -- check message content
        sys.dbms_pipe.unpack_message(vMsg);
        if(vMsg = 'stop') then
          return true;
        end if;

      end if;

    end if;

    return false;

  end;


  -- Checks if error caused by external interrupt
  function IsInterruptError(piSQLCode in number) return boolean
  is
  begin

    if( piSQLCode in (

          -1013,  -- ORA-01013: User requested cancel of current operation
          -28,    -- ORA-00028: Your session has been killed
          -13638, -- ORA-13638: The user interrupted the current operation
          -13639, -- ORA-13639: The current operation was interrupted because it timed out.
          -13668, -- ORA-13668: The current operation was aborted because it was blocking another session
          -48223, -- ORA-48223: Interrupt Requested - Fetch Aborted - Return Code [string] [string]
          -48495  -- ORA-48495: Interrupt requested

        )
      ) then

      return true;

    end if;

    return false;

  end;

  -- Main procedure for dispatcher job
  procedure MainProc
  is
    vIsFound       boolean;
    vSQLCode       number;
    vErrMsg        varchar2(2048);
    vCheckInterval number;
    vMESSAGE       SOME_CUSTOM_MESSAGE_DATA_TYPE;  -- Just for example
  begin

    vCheckInterval := NORMAL_CHECK_INTERVAL;

    while(true) loop

      vIsFound := false;

      begin

        vMESSAGE := GET_NEXT_MESSAGE_FROM_QUEUE; -- Just for example

        if(vMESSAGE is not null) then

          vIsFound := true;

          -- Process received message her
          DISPATCH_MESSAGE(vMESSAGE);  -- Just for example

          -- Commit changes to save a redo log from overflow.
          commit;

        end if;

      exception
        when others then begin

          vSQLCode := SQLCODE;
          vErrMsg := SQLERRM;

          if( IsInterruptError(vSQLCode) ) then
            -- Promote error if interrupted forcibly
            raise;
          end if;

          -- In other cases just write error conditions to log and continue.
          -- Log writing totally skipped from this example, so only comment here.

          -- Also it's a place to perform extra error analysis. 
          -- E.g. if it is some temporary error caused by remote database shutdown
          -- and so on, then increase vCheckInterval and don't stress server.

        end;
      end;

      -- Check if interrupted programmatically from control procedure
      if( CheckIsStopped ) then
        -- normal exit, job finished
        return;
      end if;

      -- If there are no new request then put process in sleep state
      -- for a short time to release resources
      if(not vIsFound) then
        dbms_lock.sleep(vCheckInterval);
      end if;

    end loop;

    -- This point never reached
    null;

  end;

end;

请注意,上面的代码是一个简化的示例(例如,没有记录执行的活动)。此外,采用示例代码来匹配问题条件并包含一些假设,如 Oracle 用户名和 GET_NEXT_MESSAGE_FROM_QUEUE 等泛型。请随时询问示例代码中是否有不清楚的地方。

【讨论】:

    猜你喜欢
    • 2017-09-14
    • 2015-08-06
    • 1970-01-01
    • 2018-05-14
    • 2018-12-26
    • 2018-02-18
    • 2020-11-27
    • 2010-09-19
    • 2012-01-19
    相关资源
    最近更新 更多