利用ORACLE JOB 模拟多线程应用

最近在做一个信息集成发布的项目，主要功能是根据用户输入的集成条件实现数据的过滤和目标数据的生成。由于数据量比较大，而且数据来源都是生产环境下的，完全实时的方式会对生成数据库造成压力，项目组考虑使用ORACLE存储过程的方式来实现，采用多个ORACLE JOB来模拟实现多线程的方式完成。

为了简单起见，前台把启动任务的相关参数信息写入到一张表，如下为表的结构：

--任务基础信息表
create table log_backstage_task (
id number,  --任务日志主键
begin_time date,--任务开始时间
end_time date,  --任务结束时间
start_type number(2),--启动类型 0 定时启动  1 人工启动
status number(1),    --任务状态 0 未启动   1 执行中 2 执行完成  3 执行失败
outline varchar2(400),--任务过程描述
task_parameter clob   --任务参数列表
);
--JOB模拟线程状态表
create table t_job_info (
job_number number , --主键  直接对应为ORACLE 的JOB编号
create_time date  , --JOB创建时间
isbusy  number(1) default 0  --是来标识JOB是否正在运行
);
--测试用数据表
create table DEMO1
(
  STR VARCHAR2(100)
);

于是根据JAVA里面的多线程编程思想，用主线程接收请求，然后看有无空闲子线程，有的话则启动子线程，让子线程接管待处理的请求。

--package 代码
create or replace package pk_full_integrate is
--定义并发处理的JOB数量   默认配置为5
  MAX_JOB_INSTANCE constant integer := 5;
--放空过程  用于子线程JOB创建后的默认执行过程
   procedure pro_null ;
--主JOB调用过程
procedure pro_job_schedule_main;
--子线程 执行过程..
procedure pro_job_execute_main(pJobNum number,pLog_id number  );
end pk_full_integrate;
/

--package body 代码
create or replace package body pk_full_integrate is
--空过程  用于子线程JOB创建后的默认执行过程   
procedure pro_null as
   begin
     null;
   end;
--判断是否还有空闲的JOB可以使用  有则返回JOB编号，没有则返回空
function fun_isHavaNullJob return number is
iResult number := null ;
begin
     select m.job_number into iResult
      from t_job_info m where m.isbusy = 0  and rownum <2; 
      return iResult;
exception when others then 
      return null ;
end;
--寻找最先提交的还没有启动的任务 有则返回ID，没有则返回空...
function fun_isHaveNotStartTask return number is
iResult number := null  ;
begin
     select t.id   into iResult  from (
     select id from log_backstage_task m where m.status = 0 order by id asc ) t
      where rownum < 2  ;
      return iResult;
exception when others then 
      return  null ;        
end;
--主JOB调用过程
procedure pro_job_schedule_main  is 
ptempJob_num number := null ;
ptempTask_num number := null; 
pExecSql varchar2(500) := ''; 
begin
     ptempJob_num := fun_isHavaNullJob;
     ptempTask_num := fun_isHaveNotStartTask ;
     while (ptempJob_num is not null ) and ( ptempTask_num is not null   )  loop
           begin
               pExecSql := 'pk_full_integrate.pro_job_execute_main('||ptempJob_num||','||ptempTask_num||' );' ;
                
               update t_job_info  m set m.isbusy = 1 where m.job_number = ptempJob_num;
               update log_backstage_task s set  s.status = 1 where s.id = ptempTask_num ;                
               --dbms_job.run(ptempJob_num);
               --改用dbms_job.broken函数，防止JOB出错之后无法重新启动，采用延迟10秒之后启动
dbms_job.what(ptempJob_num,pExecSql);
dbms_job.broken(ptempJob_num,false,sysdate+1/8640);
               commit;
               ptempJob_num := fun_isHavaNullJob;
               ptempTask_num := fun_isHaveNotStartTask ;
           exception when others then 
               dbms_output.put_line('error'||substr(sqlerrm,1,255)); 
               rollback; 
           end ;
     end loop;
 end;

--子线程 执行过程..
procedure pro_job_execute_main(pJobNum number,pLog_id number  ) is 
begin
     --dbms_output.put_line('executed successful ........ '||pJobNum);
     insert into demo1 values ('executed successful ........ '||pJobNum);
     update log_backstage_task set status = 1 where id = pLog_id ;
     update t_job_info set isBusy = 0 where job_number = pJobNum ;
     commit;
exception when others then 
     rollback;
     update log_backstage_task set status = 4  where id = pLog_id ;
     update t_job_info set isBusy = 0 where job_number = pJobNum ;
     commit;
end;
   
begin
     --包初始化脚本 
      --初始化JOB信息
     select count(1)  into pJob_cnt  from t_job_info ;
    if pJob_cnt < MAX_JOB_INSTANCE then 
        for i in 1..MAX_JOB_INSTANCE-pJob_cnt    loop
--子线程JOB的时间间隔调整为356000(1000年)，理论上不会自动执行            
dbms_job.submit( pJob_num,'pk_full_integrate.pro_null;',sysdate,'sysdate+356000');
            insert into t_job_info (JOB_NUMBER) values (pJob_num) ;
        end loop;
     end if;
     commit; 
     dbms_output.put_line('job init success......');
exception when others then 
     dbms_output.put_line('occer errors,job init failure...');
end pk_full_integrate;
/

测试代码，查看执行结果，模拟成功! 呵呵。。

executed successful ........ 125
executed successful ........ 125
executed successful ........ 125
executed successful ........ 126
executed successful ........ 126
executed successful ........ 127
executed successful ........ 127
executed successful ........ 128
executed successful ........ 128
executed successful ........ 129
executed successful ........ 129

这里用到了dbms_job包的submit,what和broken函数，用法如下：
1. Broken()过程更新一个已提交的工作的状态，典型地是用来把一个已broken的job标记为未broken工作。
这个过程有三个参数：job 、broken与next_date。

Broken (job       IN binary_integer,
                  Broken    IN boolean,
                  next_date IN date :=SYSDATE)

job参数是工作号，它在问题中唯一标识工作。
broken参数指示此工作是否将标记为broken——TRUE说明此工作将标记为停止执行，而FLASE说明此工作将标记为正常执行。
next_date参数指示在什么时候此工作将再次运行。此参数缺省值为当前日期和时间。
使用Submit()过程，工作被正常地计划好。
这个过程有五个参数：job、what、next_date、interval与no_parse。

2. Submit ( job       OUT binary_ineger,
                   What      IN  varchar2,
                   next_date IN  date,
                   interval  IN  varchar2,
                   no_parse  IN  booean:=FALSE)

job参数是由Submit()过程返回的binary_ineger。这个值用来唯一标识一个工作。
what参数是将被执行的PL/SQL代码块。
next_date参数指识何时将运行这个工作。
interval参数何时这个工作将被重执行。
no_parse参数指示此工作在提交时或执行时是否应进行语法分析——TRUE
指示此PL/SQL代码在它第一次执行时应进行语法分析，
而FALSE指示本PL/SQL代码应立即进行语法分析。

3
What()过程应许在工作执行时重新设置此正在运行的命令。这个过程接收两个参数：job与what。

PROCEDURE What (job  IN binary_ineger,
                What IN OUT varchar2)

job参数标识一个存在的工作。what参数指示将被执行的新的PL/SQL代码。


这里使用了动态SQL用来接收任务ID和JOB编号，实现动态分配，貌似在ORACLE 10G中已经对DBMS_JOB包进行了升级，有了新的DBMS_SHEDULE包，有时间再研究一下。