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

用户1148526

发布于 2019-05-25 19:38:21

6790

发布于 2019-05-25 19:38:21

文章被收录于专栏：Hadoop数据仓库

应用程序开发人员的需求是这样的：

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

设计思路：

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

源代码：

c-sharp view plain copy

-- 1. 生成可用代码池，重复执行过程为每一个国家生成一个可用代码池表，表名为org_code_加上两位国家代码
CREATE OR REPLACE PROCEDURE p_org_code
IS
BEGIN
-- tmp是一个提交后删除行的全局临时表
FOR i IN 0 .. 9999999
LOOP
INSERT INTO tmp
VALUES (LPAD (i, 7, '0'));
END LOOP;
INSERT INTO org_code_00
SELECT \*
FROM tmp
ORDER BY DBMS\_RANDOM.VALUE;
COMMIT;
END;
-- 2. 代码生成过程
CREATE OR REPLACE PROCEDURE p_gen_code (
p_country_code IN VARCHAR2, -- 国家代码
p_p1 IN VARCHAR2, -- 第一位规则字符串，如012345
p_p2 IN VARCHAR2, -- 第二位规则字符串，如012345
p_p3 IN VARCHAR2, -- 第三位规则字符串，如012345
p_p4 IN VARCHAR2, -- 第四位规则字符串，如012345
p_p5 IN VARCHAR2, -- 第五位规则字符串，如012345
p_p6 IN VARCHAR2, -- 第六位规则字符串，如012345
p_p7 IN VARCHAR2, -- 第七位规则字符串，如012345
p_instr IN VARCHAR2, -- 商品ID字符串，用,做分隔符
p_count IN NUMBER DEFAULT 1000, -- 每次提交的个数
r_outstr OUT VARCHAR2 -- 输出需要生成的代码个数和实际生成的代码个数
)
IS
l_id DBMS_SQL.varchar2_table;
l_mc DBMS_SQL.varchar2_table;
l_code DBMS_SQL.varchar2_table;
l_mc_str VARCHAR2 (255);
l_code_str VARCHAR2 (7);
l_id_count INT := 0;
l_code_count INT := 0;
l_idx INT;
l_start INT := 1;
l_substr VARCHAR2 (32);
l_instr VARCHAR2 (2000);
l_length INT;
l_tablename VARCHAR2 (30) := 'org_code_' || p_country_code;
l_sql VARCHAR2 (2000);
l_count INT := 0;
l_pointer INT := 0;
l_real_count INT := 0;
-- IPC
l_pipename CONSTANT VARCHAR2 (12) := 'mypipe';
l_pipe_getcount CONSTANT VARCHAR2 (12) := 'getcount';
l_pipe_retcount CONSTANT VARCHAR2 (12) := 'retcount';
l_send_result INT;
BEGIN
DBMS_PIPE.PURGE (l_pipename);
IF p_instr IS NULL
THEN
BEGIN
SELECT ID, mc
BULK COLLECT INTO l\_id, l\_mc
FROM product;
l\_id\_count := l\_id.COUNT;
l\_sql :=
'select code1 from (select code1 from '
|| l\_tablename
|| ' where instr('''
|| p\_p1
|| ''',substr(code1,1,1)) > 0 and instr('''
|| p\_p2
|| ''',substr(code1,2,1)) > 0 and instr('''
|| p\_p3
|| ''',substr(code1,3,1)) > 0 and instr('''
|| p\_p4
|| ''',substr(code1,4,1)) > 0 and instr('''
|| p\_p5
|| ''',substr(code1,5,1)) > 0 and instr('''
|| p\_p6
|| ''',substr(code1,6,1)) > 0 and instr('''
|| p\_p7
|| ''',substr(code1,7,1)) > 0 where rownum <= '
|| l\_id\_count;
EXECUTE IMMEDIATE l\_sql
BULK COLLECT INTO l\_code;
l\_code\_count := l\_code.COUNT;
WHILE (l\_count < l\_code\_count)
LOOP
IF DBMS\_PIPE.receive\_message (l\_pipename, 0) = 0
THEN
DBMS\_PIPE.unpack\_message (l\_pipebuf);
EXIT WHEN l\_pipebuf = 'stop';
END IF;
IF DBMS\_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);
END IF;
l\_real\_count :=
LEAST (p\_count, l\_code\_count - l\_pointer \* p\_count);
FOR i IN 1 .. l\_real\_count
LOOP
INSERT INTO product\_code
VALUES (l\_code (l\_pointer \* p\_count + i),
l\_mc (l\_pointer \* p\_count + i),
p\_country\_code);
EXECUTE IMMEDIATE 'delete from '
|| l\_tablename
|| ' where code1=:x'
USING l\_code (l\_pointer \* p\_count + i);
UPDATE product
SET status = 1
WHERE ID = l\_id (l\_pointer \* p\_count + i);
END LOOP;
COMMIT;
l\_pointer := pointer + 1;
l\_count := l\_count + l\_real\_count;
END LOOP;
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
WHEN NO\_DATA\_FOUND
THEN
COMMIT;
WHEN OTHERS
THEN
RAISE;
END;
ELSE
IF SUBSTR (p\_instr, -1, 1) = ','
THEN
l\_instr := p\_instr;
ELSE
l\_instr := p\_instr || ',';
END IF;
l\_length := LENGTH (l\_instr);
<<outer\_loop>>
WHILE l\_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;
IF DBMS\_PIPE.receive\_message (l\_pipename, 0) = 0
THEN
DBMS\_PIPE.unpack\_message (l\_pipebuf);
EXIT WHEN l\_pipebuf = 'stop';
END IF;
IF DBMS\_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);
END IF;
BEGIN
SELECT mc
INTO l\_mc\_str
FROM product
WHERE ID = l\_substr;
EXCEPTION
WHEN OTHERS
THEN
GOTO outer\_loop;
END;
l\_sql :=
'select code1 from (select code1 /*+ first_rows */ from '
|| l\_tablename
|| ' where instr('''
|| p\_p1
|| ''',
substr(code1,1,1)) > 0 and instr('''
|| p\_p2
|| ''',
substr(code1,2,1)) > 0 and instr('''
|| p\_p3
|| ''',
substr(code1,3,1)) > 0 and instr('''
|| p\_p4
|| ''',
substr(code1,4,1)) > 0 and instr('''
|| p\_p5
|| ''',
substr(code1,5,1)) > 0 and instr('''
|| p\_p6
|| ''',
substr(code1,6,1)) > 0 and instr('''
|| p\_p7
|| ''',
substr(code1,1,1)) > 0 ) where rownum=1 ';
BEGIN
EXECUTE IMMEDIATE l\_sql
INTO l\_code\_str;
EXCEPTION
WHEN NO\_DATA\_FOUND
THEN
EXIT;
END;
IF SQL%ROWCOUNT = 1
THEN
l\_code\_count := l\_code\_count + 1;
END IF;
INSERT INTO product\_code
VALUES (l\_code\_str, l\_wzmc\_str, p\_country\_code);
EXECUTE IMMEDIATE 'delete from ' || l\_tablename || ' where code1=:x'
USING l\_code\_str;
UPDATE product
SET status = 1
WHERE ID = l\_substr;
END LOOP;
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);
END IF;
END p_gen_code;
-- 3. 终止过程
CREATE OR REPLACE PROCEDURE p_stop
IS
l_pipename VARCHAR2 (12) := 'mypipe';
l_create_result INTEGER := DBMS_PIPE.create_pipe (l_pipename);
l_send_result INTEGER;
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);
END p_stop;
-- 4. 取得当前已经生成代码的个数
CREATE OR REPLACE FUNCTION fn_getcount (p_timeout NUMBER)
RETURN VARCHAR2
IS
l_pipebuf VARCHAR2 (20);
l_pipe_getcount VARCHAR2 (12) := 'getcount';
l_pipe_retcount VARCHAR2 (12) := 'retcount';
l_status NUMBER;
l_receive NUMBER;
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));
IF l_receive = 0
THEN
DBMS\_PIPE.unpack\_message (l\_pipebuf);
RETURN TO\_CHAR (l\_pipebuf);
ELSE
RETURN TO\_CHAR ('Timed out!');
END IF;
END fn_getcount;

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2016年12月28日，如有侵权请联系 cloudcommunity@tencent.com 删除

tcp/ip

面向对象编程

编程算法

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

tcp/ip

面向对象编程

编程算法

登录后参与评论

0 条评论

热度

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

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

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐