在分布式事务中实现基于Oracle PLSQL UL LOCK的悲观离线锁

背景

应用项目组每个小时会定时的run一个存储过程进行结算，每次执行的时间也许会超过一个小时，而且需要绝对保证存储过程的串行执行。因为使用内存锁不能绝对保证两个存储过程的串行执行，因为应用服务器down掉重启后可能会出现并发执行的情况，因为先前的存储过程还在db中运行。我们是使用LTS，对quartz进行了封装来做任务调度的。我们决定锁的管理操作由framework来实现。原因是：

l         锁管理器可以做成通用的模块

l         申请锁，释放锁是比较危险的操作，担心业务开发人员由于遗忘导致死锁或者并发问题

l         可以很好的集成到我们现有的framework中，方便地开放给业务开发人员使用

注意：我们极其不推荐使用悲观离线锁，如果冲突出现的概率比较少，可以用其他方法比如乐观离线锁，DB Constraint再通过补偿操作能解决的问题，请不要使用悲观离线锁。

 

原理

PLSQL UL LOCK是oracle提供出来给开发人员使用的锁资源，功能和DML锁是类似的，当然我们可以通过DML锁来完成并发控制，select…for update或者自己维护一张锁表，考虑到实现代价，我们打算使用PLSQL UL LOCK。而且Oracle保证了session释放时，UL lock都会被释放。

但是用它时，需要注意到它的DBMS_LOCK.Unique函数它每次都会commit数据。如果是在分布式事务当中，会抛出事务已提交的异常。因为我们使用的是XA resource并且transaction level是global的，也就是JTA。为了使得锁的申请和释放不影响分布式业务事务，或者我们是使用非xa的resource和local的transaction来完成锁操作，或者也可以暂停已有事务，等锁操作完成后resume暂停的分布式事务。考虑到重用已有的xa resource我们打算使用后一种方法，其实这种方法我们也会经常使用，暂停分布式事务做DDL操作，再释放事务。

 

实现方法：

l         封装DBMS_LOCK包中的几个存储过程为我们所用

l         Java端提供一个基于PLSQL UL锁的管理器

l         Java端定义好申请锁，业务操作，释放锁的使用流程，作为一个模板

 

DB存储过程：

对DBMS_LOCK做了简单的封装，避免直接调用DBMS_LOCK。这样做的好处是:

l         和Oracle解耦，如果其他数据库可以提供类似的功能，我们也可以用同名的存储过程实现

l         方便以后对存储过程重构，升级

l         我们需要对DBMS_LOCK进行简单的封装，因为DBMS_LOCK.Unique获取lockhandle oracle中锁的唯一标识，输入是lockname，逻辑名，输出是锁的唯一标识，对java端应该是透明的，java端应该只关心锁的逻辑名。

create or replace package body frm_lts_processor_lock_pkg is
   /* table to store lockhandles by name */
   TYPE handle_tbltype IS TABLE OF varchar2(128)
      INDEX BY varchar2(128);

   v_lockhandle_tbl handle_tbltype;
  
 procedure frm_lts_lock_acquire(i_lock_name in varchar2, i_expiration_time in Integer default 864000, i_wait_time in Integer default DBMS_LOCK.maxwait, o_result out number) as
      v_result     number;
      v_lockhandle varchar2(128);
   begin
   if v_lockhandle_tbl.count = 0 then
      sys.dbms_lock.allocate_unique(i_lock_name, v_lockhandle, i_expiration_time);
      v_lockhandle_tbl(i_lock_name) := v_lockhandle;
   elsif v_lockhandle_tbl.exists(i_lock_name) then
      dbms_output.put_line('atttacked');
      v_lockhandle := v_lockhandle_tbl(i_lock_name);
   else
     dbms_output.put_line('new acquire');
      --acquire a unique lock id
     sys.dbms_lock.allocate_unique(i_lock_name, v_lockhandle, i_expiration_time);
     v_lockhandle_tbl(i_lock_name) := v_lockhandle;
       
   end if;
    --acquire a lock
    v_result := sys.dbms_lock.request(v_lockhandle, dbms_lock.x_mode, i_wait_time, false);
 
    --set return values
     o_result := v_result;
 end frm_lts_lock_acquire;
 
 function frm_lts_lock_release(i_lock_name in varchar2) return number as
       v_result number := 6;
       v_lockhandle varchar2(128);
 begin
      --release lock according to lockhandle
      if v_lockhandle_tbl.exists(i_lock_name) then
          v_lockhandle :=  v_lockhandle_tbl(i_lock_name);
          v_result := sys.dbms_lock.release(v_lockhandle);
          v_lockhandle_tbl.delete(i_lock_name);       
      end if;
      return v_result;
 end frm_lts_lock_release;

end frm_lts_processor_lock_pkg;
/

 

 

锁管理器：

其实应用项目组有多个这样的存储过程，而这些存储过程之间的串行执行可以有多个business key来决定的，比如job order number，delivery order等。所以我们需要给他们提供多锁管理机制。我们会对这多个锁进行排序，以避免死锁，并强烈推荐应用项目设置超时时间。这些business key是由String对象构成的，为了防止大量的业务操作被锁在null或者空string这样没有意义的business key上面，我们对application提供的锁集合还需要进行过滤。

原理还是很简单的，就是在本地事务中调用db端的申请锁，释放锁的存储过程，然后对返回的结果进行一系列处理。

在使用多锁机制的时候要保证，如果只申请到了部分锁，在申请其中另外一个锁时发生了错误或者超时，要能够安全地将已申请的锁释放掉，所以多锁申请需要记录已申请到的锁，并且记录发生的错误，区分timeout和异常。Timeout返回false，如果出现异常记录下来，最后抛出。释放多锁时，不能被中断，记录释放每个锁后的结果，最后判定如果其中一些锁释放时发生了错误，抛出。

 

handleLock定义暂停jta事务，执行锁操作，释放jta事务流程

 

private Object handleLock(Connection connection,
            LocalTransactionCallback localTransactionCallback)
            throws LockException {
        TransactionManager tm = null;
        Transaction currentTx = null;
        Object result = null;
        try {
            Context initialContext = new InitialContext();
            UserTransaction userTrx = (javax.transaction.UserTransaction) initialContext
                    .lookup("java:comp/UserTransaction");
            if (!(userTrx.getStatus() == Status.STATUS_NO_TRANSACTION)) {
                tm = TransactionUtils.getTransactionManager(userTrx);
                if (tm != null) {
                    currentTx = tm.suspend();
                }
            }
            result = localTransactionCallback
                    .executeInLocalTransaction(connection);

            if (null != currentTx) {
                tm.resume(currentTx);
            }
        } catch (NamingException e) {
        } catch (SystemException e) {
        } catch (InvalidTransactionException e) {
        } catch (IllegalStateException e) {
        }
        return result;
    }

 多锁申请操作是上面流程的一个回调

 

private class ObtainMutipleLocksLocalTransactionCallback implements
            LocalTransactionCallback {
        private Set<String> lockNames;
        private int waitTime;

        ObtainMutipleLocksLocalTransactionCallback(Set<String> lockNames,
                int waitTime) {
            this.lockNames = lockNames;
            this.waitTime = waitTime;
        }
        public Object executeInLocalTransaction(Connection conn) {
            CallableStatement lockAcquireStmt = null;
            Set<String> obtainedLockNames = new HashSet<String>();
            boolean timeOut = false;
            String timeOutLockName = null;
            Exception mifLockException = null;
            try {
                lockAcquireStmt = conn.prepareCall(OBTAIN_LOCK_PROC_CALL);
                for (String lockName : lockNames) {
                    lockAcquireStmt.setString(1, lockName);
                    lockAcquireStmt.setInt(2, LCOK_EXPIRE_TIME);
                    lockAcquireStmt.setInt(3, waitTime);
                    lockAcquireStmt.registerOutParameter(4,
                            java.sql.Types.INTEGER);
                    lockAcquireStmt.registerOutParameter(5,
                            java.sql.Types.VARCHAR);
                    lockAcquireStmt.executeUpdate();
                    int lockacquireResult = lockAcquireStmt.getInt(4);
                    if (lockacquireResult == ULLockResultType.SUCCESSFUL) 
                        obtainedLockNames.add(lockName);
                    } else if (lockacquireResult == ULLockResultType.TIMEOUT) {
                        timeOut = true;
                        timeOutLockName = lockName;
                        break;
                    } else if (lockacquireResult != ULLockResultType.ALREADY_OWNED) {
                        String lockResultDesc = ULLockResultType
                                .getAcquireTypeDesc(lockacquireResult);
                        LockException lockException = new LockException(
                                "Obtain lock " + lockName
                                        + " fails, the reason is "
                                        + lockResultDesc + " .");
                        lockException.setLockName(lockName);
                        lockException.setLockHandlingResult(lockResultDesc);
                        throw lockException;
                    } else {
                    }
                }
            } catch (Exception ex) {
                mifLockException = ex;
            } finally {
                if (null != lockAcquireStmt) {
                    try {
                        lockAcquireStmt.close();
                    } catch (SQLException e) {
                        // swallow
                    }
                }
            }
            boolean success = true;
            if (timeOut || mifLockException != null) {
                success = false;
            }
            return new ObtainMultipleLocksResult(success, obtainedLockNames,
                    timeOut, timeOutLockName, mifLockException);
        }
    }

 

多锁释放操作也是事务暂停流程的一个回调

 

 

private class ReleaseMultipleLocksLocalTransactionCallback implements
            LocalTransactionCallback {
        private Set<String> lockNames;

        ReleaseMultipleLocksLocalTransactionCallback(Set<String> lockNames) {
            this.lockNames = lockNames;
        }

        public Object executeInLocalTransaction(Connection conn) {
            CallableStatement lockReleaseStmt = null;
            Map<String, Exception> mifLockErrors = new HashMap<String, Exception>();
            Set<String> releasedLocks = new HashSet<String>();
            try {
                try {
                    lockReleaseStmt = conn.prepareCall(RELEASE_LOCK_PROC_CALL);
                } catch (Exception ex) {
                    for (String lockName : lockNames) {
                        mifLockErrors.put(lockName, ex);
                    }
return new ReleaseMutipleLocksResult(false, releasedLocks, mifLockErrors);
                }

                for (String lockName : lockNames) {
                    try {
                        lockReleaseStmt.registerOutParameter(1,
                                java.sql.Types.INTEGER);
                        lockReleaseStmt.setString(2, lockName);
                        lockReleaseStmt.executeUpdate();
                        int lockReleaseResult = lockReleaseStmt.getInt(1);
                        if (lockReleaseResult == ULLockResultType.SUCCESSFUL) {
                            releasedLocks.add(lockName);
                        } else {
                            String lockResultDesc = ULLockResultType
                                    .getReleaseTypeDesc(lockReleaseResult);
                            LockException lockException = new LockException(
                                    "Release lock " + lockName
                                            + " fails, the reason is "
                                            + lockResultDesc + " .");
                            lockException.setLockName(lockName);
                            lockException.setLockHandlingResult(lockResultDesc);
                            mifLockErrors.put(lockName, lockException);
                        }
                    } catch (Exception ex) {
                        mifLockErrors.put(lockName, ex);
                    }
                }
            } finally {
                if (null != lockReleaseStmt) {
                    try {
                        lockReleaseStmt.close();
                    } catch (SQLException e) {
                    }
                }
            }
            boolean success = releasedLocks.size() == this.lockNames.size();
            return new ReleaseMutipleLocksResult(success, releasedLocks,
                    mifLockErrors);
        }
    }

 

使用模板：注意锁的释放要写在finally语句块里面，保证锁的释放。

定义好模板，防止Application用户直接调用锁管理器或者滥用锁，忘记释放锁。我们决定定义一个模板，做到锁的申请和释放对application用户来说是透明的，把它做成了隐含锁。

  

public void execute(JobExecutionContext context)
            throws JobExecutionException {
        Map jobDataMap = context
        .getJobDetail().getJobDataMap();
        Collection<String> lockKeys = (Collection<String>) jobDataMap.get(LOCK_NAME_KEY);
        Integer waitTimeInteger = (Integer) jobDataMap
        .get(LOCK_WAIT_TIME_SECONDS_KEY);
        int waitTime = MAX_WAITTIME;
        if (waitTimeInteger != null) {
            waitTime = waitTimeInteger.intValue();
        }
        Set<String> uniqueLockKeys = new HashSet<String>(lockKeys);

        // filter empty keys
        Iterator<String> keyIterator = uniqueLockKeys.iterator();
        while (keyIterator.hasNext()) {
            String key = keyIterator.next();
            if (StringUtils.isEmptyNoOffset(key)) {
                keyIterator.remove();
            }
        }
        if (CollectionUtils.isNotEmptyCollection(uniqueLockKeys)) {
            Set<String> obtainedLockNames = null;
            Connection connection = null;
            try {
                connection = DataSource.getConnection();
                ObtainMultipleLocksResult result = LOCK_MANAGER.obtainLock(
                        connection, uniqueLockKeys, waitTime);
                obtainedLockNames = result.getObtainedLockNames();
                if (!result.isSuccess()) {
                    if (result.isTimeout()) {
                      //do log
                        return;
                    } else {
                        JobExecutionException jobException = new JobExecutionException(
                                "Obtain locks failed! "
                                        + result.getMifLockException()
                                                .getMessage(), result
                                        .getMifLockException());
                        throw jobException;
                    }
                }
                this. executeInLock (context);
            } catch (Throwable e) {
                throw new JobExecutionException(
                        "Get db connection failed!" + e.getMessage(), e);
            } finally {
                if (null != connection) {
                    this.releaseLocks(connection, obtainedLockNames);
                    try {
                        connection.close();
                    } catch (SQLException e) {
                        throw new JobExecutionException(
                                "close  db connection failed!" + e.getMessage(), e);
                    }
                }
            }
        } else {
            this.executeInLock(context);
        }
    }

 

executeInLock由application的子类继承实现

 

缓存

l         缓存悲观离线锁

l         缓存lockhandle

因为使用的是悲观离线锁，每次申请锁都要跑一趟db，但如果当前线程已经是lock的所有者就不需要白跑一趟了。可以用ThreadLocal把当前线程已经拥有的锁缓存起来，释放锁时对应的需要清除缓存。

在申请锁时，需要获得UL Lock时的lockhandle，释放锁时也需要提供锁的lockhandle，我们需要将它缓存起来，主要是因为DBMS_LOCK.Unique每次都会commit，会影响性能，这样每次释放锁时就可以直接使用lockhandle了。有两种方法对lockhandle进行缓存，缓存在java端作为实例变量，缓存在plsql包的全局变量中。缓存在java端需要注意的是，lock manager不能作为单例或者享元来使用，否则lock handle的缓存在多jvm之间也存在着并发控制和同步的问题。

 

源代码： 见附件

参考：

http://docstore.mik.ua/orelly/oracle/bipack/ch04_01.htm