利用Oracle dbms_pipe实现存储过程之间的通信

应用程序开发人员的需求是这样的：
1. 根据条件给每一个国家的商品生成唯一7位随机代码，不同国家之间的商品代码可以相同
2. 如果输入标准分隔符的字符串，则解析该字符串作为需要生成的商品ID，为其生成代码，否则为商品表中所有商品ID生成代码
3. 代码的每一位要符合相应的规则，例如第一位的规则是[0123]，则这位只能是0、1、2、3中的一个数
4. 由于可能一次生成大量的代码，这个过程需要较长时间，所以需要用进度条提示生成进度
5. 可以在生成过程执行中终止过程
6. 返回需要生成的代码个数和实际生成的代码个数
7. 每次提交的个数可以通过参数定义，例如共要生成10万的代码，每次提交1000个

设计思路：
1. 为每个国家预生成0到9999999一千万个随机数作为候选代码池表，每生成一个代码就从代码池中删除一个，避免查重操作
2. 用一个存储过程生成代码，另一个过程用来终止生成过程，两个过程间用dbms_pipe进行通信
3. 用一个函数返回当前已经生成的代码个数，供显示进度条的外部程序调用，生成代码过程与该函数用dbms_pipe进行通信
4. 用bulk collect批量处理提高效率

源代码：

--1.生成可用代码池，重复执行过程为每一个国家生成一个可用代码池表，表名为org_code_加上两位国家代码
CREATEORREPLACEPROCEDUREp_org_code
IS
BEGIN
--tmp是一个提交后删除行的全局临时表
FORiIN0..9999999
LOOP
INSERTINTOtmp
VALUES(LPAD(i,7,'0'));
ENDLOOP;

INSERTINTOorg_code_00
SELECT*
FROMtmp
ORDERBYDBMS_RANDOM.VALUE;

COMMIT;
END;

--2.代码生成过程
CREATEORREPLACEPROCEDUREp_gen_code(
p_country_codeINVARCHAR2,--国家代码
p_p1INVARCHAR2,--第一位规则字符串，如012345
p_p2INVARCHAR2,--第二位规则字符串，如012345
p_p3INVARCHAR2,--第三位规则字符串，如012345
p_p4INVARCHAR2,--第四位规则字符串，如012345
p_p5INVARCHAR2,--第五位规则字符串，如012345
p_p6INVARCHAR2,--第六位规则字符串，如012345
p_p7INVARCHAR2,--第七位规则字符串，如012345
p_instrINVARCHAR2,--商品ID字符串，用,做分隔符
p_countINNUMBERDEFAULT1000,--每次提交的个数
r_outstrOUTVARCHAR2--输出需要生成的代码个数和实际生成的代码个数
)
IS
l_idDBMS_SQL.varchar2_table;
l_mcDBMS_SQL.varchar2_table;
l_codeDBMS_SQL.varchar2_table;
l_mc_strVARCHAR2(255);
l_code_strVARCHAR2(7);
l_id_countINT:=0;
l_code_countINT:=0;
l_idxINT;
l_startINT:=1;
l_substrVARCHAR2(32);
l_instrVARCHAR2(2000);
l_lengthINT;
l_tablenameVARCHAR2(30):='org_code_'||p_country_code;
l_sqlVARCHAR2(2000);
l_countINT:=0;
l_pointerINT:=0;
l_real_countINT:=0;
--IPC
l_pipenameCONSTANTVARCHAR2(12):='mypipe';
l_pipe_getcountCONSTANTVARCHAR2(12):='getcount';
l_pipe_retcountCONSTANTVARCHAR2(12):='retcount';
l_send_resultINT;
BEGIN
DBMS_PIPE.PURGE(l_pipename);

IFp_instrISNULL
THEN
BEGIN
SELECTID,mc
BULKCOLLECTINTOl_id,l_mc
FROMproduct;

l_id_count:=l_id.COUNT;
l_sql:=
'selectcode1from(selectcode1from'
||l_tablename
||'whereinstr('''
||p_p1
||''',substr(code1,1,1))>0andinstr('''
||p_p2
||''',substr(code1,2,1))>0andinstr('''
||p_p3
||''',substr(code1,3,1))>0andinstr('''
||p_p4
||''',substr(code1,4,1))>0andinstr('''
||p_p5
||''',substr(code1,5,1))>0andinstr('''
||p_p6
||''',substr(code1,6,1))>0andinstr('''
||p_p7
||''',substr(code1,7,1))>0whererownum<='
||l_id_count;

EXECUTEIMMEDIATEl_sql
BULKCOLLECTINTOl_code;

l_code_count:=l_code.COUNT;

WHILE(l_count<l_code_count)
LOOP
IFDBMS_PIPE.receive_message(l_pipename,0)=0
THEN
DBMS_PIPE.unpack_message(l_pipebuf);
EXITWHENl_pipebuf='stop';
ENDIF;

IFDBMS_PIPE.receive_message(l_pipe_getcount,0)=0
THEN
DBMS_PIPE.PURGE(l_pipe_retcount);
dbms_pipe_pack_message(TO_CHAR(l_count)||'|running');
l_send_result:=DBMS_PIPE.send_message(l_pipe_retcount);
ENDIF;

l_real_count:=
LEAST(p_count,l_code_count-l_pointer*p_count);

FORiIN1..l_real_count
LOOP
INSERTINTOproduct_code
VALUES(l_code(l_pointer*p_count+i),
l_mc(l_pointer*p_count+i),
p_country_code);

EXECUTEIMMEDIATE'deletefrom'
||l_tablename
||'wherecode1=:x'
USINGl_code(l_pointer*p_count+i);

UPDATEproduct
SETstatus=1
WHEREID=l_id(l_pointer*p_count+i);
ENDLOOP;

COMMIT;
l_pointer:=pointer+1;
l_count:=l_count+l_real_count;
ENDLOOP;

DBMS_PIPE.PURGE(l_pipe_retcount);
DBMS_PIPE.pack_message(TO_CHAR(l_count)||'|end');
l_send_result:=DBMS_PIPE.send_message(l_pipe_retcount);
r_outstr:=l_id_count||'|'||l_count;
DBMS_OUTPUT.put_line(r_outstr);
EXCEPTION
WHENNO_DATA_FOUND
THEN
COMMIT;
WHENOTHERS
THEN
RAISE;
END;
ELSE
IFSUBSTR(p_instr,-1,1)=','
THEN
l_instr:=p_instr;
ELSE
l_instr:=p_instr||',';
ENDIF;

l_length:=LENGTH(l_instr);

<<outer_loop>>
WHILEl_start<l_length
LOOP
l_id_count:=l_id_count+1;
l_idx:=INSTR(l_instr,',',l_start);
l_substr:=SUBSTR(l_instr,l_start,l_idx-l_start);
l_start:=l_idx+1;

IFDBMS_PIPE.receive_message(l_pipename,0)=0
THEN
DBMS_PIPE.unpack_message(l_pipebuf);
EXITWHENl_pipebuf='stop';
ENDIF;

IFDBMS_PIPE.receive_message(l_pipe_getcount,0)=0
THEN
DBMS_PIPE.PURGE(l_pipe_retcount);
DBMS_PIPE.pack_message(l_code_count||'|running');
l_send_result:=DBMS_PIPE.send_message(l_pipe_retcount);
ENDIF;

BEGIN
SELECTmc
INTOl_mc_str
FROMproduct
WHEREID=l_substr;
EXCEPTION
WHENOTHERS
THEN
GOTOouter_loop;
END;

l_sql:=
'selectcode1from(selectcode1/*+first_rows*/from'
||l_tablename
||'whereinstr('''
||p_p1
||''',
substr(code1,1,1))>0andinstr('''
||p_p2
||''',
substr(code1,2,1))>0andinstr('''
||p_p3
||''',
substr(code1,3,1))>0andinstr('''
||p_p4
||''',
substr(code1,4,1))>0andinstr('''
||p_p5
||''',
substr(code1,5,1))>0andinstr('''
||p_p6
||''',
substr(code1,6,1))>0andinstr('''
||p_p7
||''',
substr(code1,1,1))>0)whererownum=1';

BEGIN
EXECUTEIMMEDIATEl_sql
INTOl_code_str;
EXCEPTION
WHENNO_DATA_FOUND
THEN
EXIT;
END;

IFSQL%ROWCOUNT=1
THEN
l_code_count:=l_code_count+1;
ENDIF;

INSERTINTOproduct_code
VALUES(l_code_str,l_wzmc_str,p_country_code);

EXECUTEIMMEDIATE'deletefrom'||l_tablename||'wherecode1=:x'
USINGl_code_str;

UPDATEproduct
SETstatus=1
WHEREID=l_substr;
ENDLOOP;

COMMIT;
DBMS_PIPE.PURGE(l_pipe_retcount);
DBMS_PIPE.pack_message(TO_CHAR(l_code_count)||'|end');
l_send_result:=DBMS_PIPE.send_message(l_pipe_retcount);
r_outstr:=
LENGTH(l_instr)
-LENGTH(REPLACE(l_instr,',',''))
||'|'
||l_code_count;
DBMS_OUTPUT.put_line(r_outstr);
ENDIF;
ENDp_gen_code;

--3.终止过程
CREATEORREPLACEPROCEDUREp_stop
IS
l_pipenameVARCHAR2(12):='mypipe';
l_create_resultINTEGER:=DBMS_PIPE.create_pipe(l_pipename);
l_send_resultINTEGER;
BEGIN
DBMS_PIPE.PURGE(l_pipename);
DBMS_PIPE.pack_message('stop');
l_send_result:=DBMS_PIPE.send_message(l_pipename);
DBMS_OUTPUT.put_line('l_send_result:'||l_send_result);
ENDp_stop;

--4.取得当前已经生成代码的个数
CREATEORREPLACEFUNCTIONfn_getcount(p_timeoutNUMBER)
RETURNVARCHAR2
IS
l_pipebufVARCHAR2(20);
l_pipe_getcountVARCHAR2(12):='getcount';
l_pipe_retcountVARCHAR2(12):='retcount';
l_statusNUMBER;
l_receiveNUMBER;
BEGIN
DBMS_PIPE.PURGE(l_pipe_getcount);
DBMS_PIPE.pack_message(l_pipe_getcount);
l_status:=DBMS_PIPE.send_message(l_pipe_getcount);
l_receive:=
DBMS_PIPE.receive_message(l_pipe_retcount,NVL(p_timeout,0));

IFl_receive=0
THEN
DBMS_PIPE.unpack_message(l_pipebuf);
RETURNTO_CHAR(l_pipebuf);
ELSE
RETURNTO_CHAR('Timedout!');
ENDIF;
ENDfn_getcount;