用批量绑定大幅提高性能的一则案例-技术开发专区

用批量绑定大幅提高性能的一则案例

作者：IT168 hanson 编辑：魏国亮 2006-09-06 11:10

【IT168 技术文档】1.案例起因

公司有两套系统，一套是供应链系统，另一套是物流系统。这两套系统分别是由两家供应商提供的。基于的数据库平台都是oracle 9i。它们之间通过一个独立的小程序进行数据交互工作，该小程序由我们项目组自行开发，我们称之为数据交互程序，该数据交互程序每隔一定时间启动。物流系统运行过程中，会不断的把它内部产生的数据插入到一个接口表里，而只要该数据交互程序启动，就会读取该接口表里的数据并进行处理，从而修改供应链系统里的数据。处理完毕以后，将处理成功的数据插入一个历史表并更新接口表里该记录的状态为成功，而没有成功的也要更新接口表里该记录的状态为失败。

刚开始，数据量还不是很大，所以数据交互程序运行一次所花的时间并不多。但是运行了半年以后，该数据交互程序运行越来越慢，发展到后来运行一次需要30分钟。我仔细检查了该交互程序以后，发现该程序主要调用一个后台存储过程，该存储过程逻辑上说比较简单，只不过是打开一个cursor，获取接口表里所有符合条件的数据，然后对每一条数据都调用另外一个存储过程进行处理，处理成功则插入历史表并设置成功标记，否则设置失败标记。于是，想到可以使用批量绑定改写一下该存储过程，看看效果。改写完以后，发现运行效率大大提高，现在运行一次只需要10分钟。下面就详细介绍一下如何使用批量绑定。

2.批量绑定的含义

oracle从8i开始引入了一个新的PL/SQL的技术：批量绑定(BULK BINDING)。原来需要很多单独的重复的SELECT、INSERT、UPDATE或DELETE的语句才能完成的功能，包括从表里获取数据和把数据插入表等，现在可以成批的一次性完成了，从而极大的提高了性能。

当运行PL/SQL时，由PL/SQL引擎将代码中的SQL语句传递给SQL引擎，从而由SQL引擎解析并执行SQL，然后再由SQL引擎将数据结果返回给PL/SQL引擎。PL/SQL引擎和SQL引擎之间的数据交互叫做上下文切换。而批量绑定能够提高性能的本质就在于它极大降低了上下文切换的次数。

为了实现批量绑定，oracle提供了两种方法：

使用FORALL为INSERT、UPDATE或DELETE提供批量绑定；
使用BULK COLLECT为SELECT提供批量绑定。

BULK COLLECT在SQL引擎将数据结果返回给PL/SQL引擎前，把输出结果批量绑定到一个结果集里。然后再一次性的传递给PL/SQL引擎。可以在SELECT INTO、FETCH INTO和RETURNING INTO里使用。语法结构为：

[SELECT INTO|FETCH INTO|RETURNING INTO]
BULK COLLECT INTO collection_name[,collection_name]...

FORALL则是在PL/SQL引擎将DML的数据传递给SQL引擎之前，把输入数据批量绑定到一个结果集里。然后再一次性的传递给SQL引擎。语法结构为：

FORALL index IN lower_bound..upper_boundsql_statement;

尽管FORALL含有循环的机制，但是它并不是一个FOR循环。SQL引擎对每一个index都会执行一遍sql_statement。顺带提一下，9i下对于FORALL的批量绑定来说有个问题，就是如果集合的lower_bound和upper_bound之间如果不是连续的，而是存在跳号则会报如下的错：

ORA-22160: element at index [N] does not exist

比如某个集合共有4个元素，其下标分别为1、2、4、5，则如果使用FORALL将报22160的错误。这个问题在10g中通过引入INDICES OF和VALUES OF，从而得到了很好的解决。INDICES OF用在被绑定的集合是不连续的情况下。比如：

FORALL index IN INDICES OF scatter_array1
INSERT INTO t(col1,col2...colN)
VALUES (scatter_array1(index),scatter_arrary2(index)...scatter_arraryN(index));

这样，就不用担心scatter_array1里的下标是否是连续的了。

VALUES OF用在被绑定的集合是不连续的，但是在sql_statement里所引用的集合则是连续的。可以理解为使用一个不连续的指针数组，该数组里的值都是指针，这些指针指向一个连续的数组，通过将这些指针作为该连续数组里的元素下标，就可以获得该连续数组里的指定元素的值了。比如：

FORALL index IN INDICES OF scatter_pointer_array
INSERT INTO t(col1,col2...colN)
VALUES (continue_array1(index),continue_arrary2(index)...continue_arraryN(index));

3.案例说明

由于原来的存储过程因为涉及到了很多表，比较复杂，不便于读者做测试。所以我简化了该过程，但是意思是一样的。简单说来，就是创建一个接口表，取名为：pix_trans，再创建一个历史表。将数据插入该接口表。然后写一个存储过程，取出该表里的数据并逐条处理。

1)创建pix_trans

SQL> create table pix_trans(id number not null,pixtype varchar2(10),actcode
varchar2(10),status varchar2(2));
SQL> alter table pix_trans add constraint pk_pix_trans primary key(id);
SQL> create table pix_trans_his as select * from pix_trans where 1=2;

2)插入测试数据

SQL> execute dbms_random.seed(0);
SQL> insert into pix_trans select rownum,trunc(10*dbms_random.normal),
trunc(10*dbms_random.normal),'00' from dba_objects;
SQL> commit;
SQL> select count(*) from pix_trans;
COUNT(*)
 ----------
109564

可以看到，我们的测试表大概有10万多条数据。

3.1 原存储过程

    CREATE OR REPLACE PROCEDURE P_HANDLE_PIX(
    ai_rtd out number,
    as_rmsg out varchar2)
    IS
    CURSOR pix_trans
    IS
    SELECT id,pixtype,actcode
    FROM pix_trans
    WHERE status='00';
    ll_begin number;
    ll_end number;
    BEGIN
    SELECT to_char(sysdate,'sssss') INTO ll_begin FROM dual;
    --打开游标
    FOR pix IN pix_trans
    LOOP
    --实际处理每条记录
    P_HANDLE_STOCK(pix.id,pix.pixtype,pix.actcode,ai_rtd,as_rmsg);
    IF ai_rtd>0 THEN--如果成功则插入历史表，并设置成功标记
    UPDATE pix_trans SET status = '10' WHERE id = pix.id;

    INSERT INTO pix_trans_his(id,pixtype,actcode)
    VALUES (pix.id,pix.pixtype,pix.actcode);
    ELSE--如果失败，则设置失败标记
    UPDATE pix_trans SET status = '90' WHERE id = pix.id;
    END IF;
    END LOOP;
    SELECT to_char(sysdate,'sssss') INTO ll_end FROM dual;
    DBMS_OUTPUT.PUT_LINE('for loop:' || to_char(ll_end - ll_begin));
    END;

这个存储过程的逻辑比较简单，其中，P_HANDLE_STOCK是实际商务逻辑的处理的过程，读者可以自己随便写，我这里写了一个测试的过程，简单的将pix_trans表的id被3除后余1则认为失败，否则成功。

    CREATE OR REPLACE PROCEDURE P_HANDLE_STOCK(
    ai_pixid in number,
    as_pixtype in pix_trans.pixtype%type,
    as_pixcode in pix_trans.actcode%type,
    ai_rtd out number,
    as_rmsg out varchar2
    )
    IS
    BEGIN
    IF mod(ai_pixid,3) = 1 THEN
    ai_rtd := -1;
    as_rmsg := 'test';
    ELSE
    ai_rtd := 1;
    as_rmsg := 'done';
    END IF;
    END;

我们开始测试，看看运行一次P_HANDLE_PIX需要多长时间。

    SQL> set serveroutput on;
    SQL> declare
    2li_rtd number;
    3ls_rmsg varchar2(255);
    4begin
    5p_handle_pix(li_rtd,ls_rmsg);
    6end;
    7/
    for loop:20

    PL/SQL procedure successfully completed.

从结果看，运行一次需要20秒。

3.2改写后的存储过程（基于9i）

现在，笔者使用批量绑定的方式，也就是使用BULK COLLECT和FORALL的方式来改写上面的存储过程：P_HANDLE_PIX。

    CREATE OR REPLACE PROCEDURE P_HANDLE_PIX_NEW(
    ai_rtd out number,
    as_rmsg out varchar2)
    IS
    TYPEt_pixid IS TABLE OF pix_trans.id%TYPE INDEX BY PLS_INTEGER;
    TYPEt_pixtype IS TABLE OF pix_trans.pixtype%TYPE INDEX BY PLS_INTEGER;
    TYPEt_actcode IS TABLE OF pix_trans.actcode%TYPE INDEX BY PLS_INTEGER;

    pix_id t_pixid;
    pix_pixtype t_pixtype;
    pix_actcode t_actcode;

    succ_pix_id t_pixid;
    succ_pix_pixtype t_pixtype;
    succ_pix_actcode t_actcode;

    fail_pix_id t_pixid;

    ll_begin number;
    ll_end number;

    indx_succ number;
    indx_fail number;
    BEGIN
    SELECT to_char(sysdate,'sssss') INTO ll_begin FROM dual;
    --获得所有的pix记录
    SELECT id,pixtype,actcode
    BULK COLLECT INTO pix_id,pix_pixtype,pix_actcode
    FROM pix_trans
    WHERE status = '00';
    --如果集合为空，则退出
    IF pix_id.count=0 THEN
    ai_rtd := 0;
    as_rmsg := 'No recoreds';
    RETURN;
    END IF;
    --逐条处理记录
    indx_succ := 1;
    indx_fail := 1;
    FOR indx IN pix_id.first .. pix_id.last
    LOOP
    P_HANDLE_SOTCK(pix_id(indx),pix_pixtype(indx),pix_actcode(indx),ai_rtd,as_rmsg);
    IF ai_rtd>0 THEN
    succ_pix_id(indx_succ) := pix_id(indx);
    succ_pix_pixtype(indx_succ) := pix_pixtype(indx);
    succ_pix_actcode(indx_succ) := pix_actcode(indx);
    indx_succ := indx_succ + 1;
    ELSE
    fail_pix_id(indx_fail) := pix_id(indx);
    indx_fail := indx_fail + 1;
    END IF;
    END LOOP;
    --将处理成功的记录插入历史表
    FORALL indx IN succ_pix_id.first .. succ_pix_id.last
    INSERT INTO pix_trans_his(id,pixtype,actcode)
    VALUES (succ_pix_id(indx),succ_pix_pixtype(indx),succ_pix_actcode(indx));

    --更新记录的成功标志
    FORALL indx IN succ_pix_id.first .. succ_pix_id.last
    UPDATE pix_trans SET status = '10' WHERE id = succ_pix_id(indx);

    --更新记录的失败标志
    FORALL indx IN fail_pix_id.first .. fail_pix_id.last
    UPDATE pix_trans SET status = '10' WHERE id = fail_pix_id(indx);

    SELECT to_char(sysdate,'sssss') INTO ll_end FROM dual;
    DBMS_OUTPUT.PUT_LINE('forall:' || to_char(ll_end - ll_begin));
    END;

我们看到，使用批量绑定的后果就是程序的可读性差了，程序代码更加复杂了。我们可以再次进行测试，看看改写后运行一次需要多长时间。

   SQL> set serveroutput on;
    SQL> declare
    2li_rtd number;
    3ls_rmsg varchar2(255);
    4begin
    5p_handle_pix_new(li_rtd,ls_rmsg);
    6end;
    7/
    for loop:11

    PL/SQL procedure successfully completed.

可以看到现在运行一次只需要11秒。时间相比不使用批量绑定来说，缩短了9秒，也就是大约提高了45%的性能。这应该算是一个很惊人的数字了。还有没有办法改进呢？我们试试10g下是不是还可以更快。

3.3进一步改写存储过程（基于10g）

    CREATE OR REPLACE PROCEDURE P_HANDLE_PIX_NEW_10(
    ai_rtd out number,
    as_rmsg out varchar2)
    IS
    TYPE t_pix_pointer IS TABLE OF PLS_INTEGER;
    TYPE t_pixid IS TABLE OF pix_trans.id%TYPE INDEX BY PLS_INTEGER;
    TYPE t_pixtype IS TABLE OF pix_trans.pixtype%TYPE INDEX BY PLS_INTEGER;
    TYPE t_actcode IS TABLE OF pix_trans.actcode%TYPE INDEX BY PLS_INTEGER;

    pix_id t_pixid;
    pix_pixtype t_pixtype;
    pix_actcode t_actcode;

    succ_pix_id t_pix_pointer := t_pix_pointer();
    fail_pix_id t_pix_pointer := t_pix_pointer();

    ll_begin number;
    ll_end number;

    BEGIN
    SELECT to_char(sysdate,'sssss') INTO ll_begin FROM dual;
    --获得pix信息
    SELECT id,pixtype,actcode
    BULK COLLECT INTO pix_id,pix_pixtype,pix_actcode
    FROM pix_trans
    WHERE status = '00';
    --如果集合为空，则退出
    IF pix_id.count=0 THEN
    ai_rtd := 0;
    as_rmsg := 'No recoreds';
    return;
    END IF;
    --逐条处理记录
    FOR indx IN pix_id.first .. pix_id.last
    LOOP
    P_HANDLE_STOCK(pix_id(indx),pix_pixtype(indx),pix_actcode(indx),ai_rtd,as_rmsg);
    IF ai_rtd>0 THEN
    succ_pix_id.extend;
    succ_pix_id(succ_pix_id.last) := indx;
    ELSE
    fail_pix_id.extend;
    fail_pix_id(fail_pix_id.last) := indx;
    END IF;
    END LOOP;
    --将处理成功的记录插入历史表
    FORALL indx IN VALUES OF succ_pix_id
    INSERT INTO pix_trans_his(id,pixtype,actcode)
    VALUES (pix_id(indx),pix_pixtype(indx),pix_actcode(indx));

    --更新记录的成功标志
    FORALL indx IN VALUES OF succ_pix_id
    UPDATE pix_trans SET status = '10' WHERE id = pix_id(indx);

    --更新记录的失败标志
    FORALL indx IN VALUES OF fail_pix_id
    UPDATE pix_trans SET status = '10' WHERE id = pix_id(indx);

    SELECT to_char(sysdate,'sssss') INTO ll_end FROM dual;
    DBMS_OUTPUT.PUT_LINE('forall values of:' || to_char(ll_end - ll_begin));
    END;

在10g里，我们可以减少了两个集合，而只是引入一个失败的集合指针和一个成功的集合指针。通过失败的集合指针，我们可以找到失败的pix_trans.id，从而更新pix_trans表里对应的记录的状态；通过成功的集合指针，我们可以找到需要插入历史表的三个字段的值。

那么这样改写以后的效果如何呢？我们来测试一下运行速度。

    SQL> set serveroutput on;
    SQL> declare
    2li_rtd number;
    3ls_rmsg varchar2(255);
    4begin
    5p_handle_pix_new_10(li_rtd,ls_rmsg);
    6end;
    7/
    forall values of:10

    PL/SQL procedure successfully completed.

可以看到，相比9i下的写法，速度提高了1秒钟。大约是9%左右。效果并不是很明显。