集群环境下防止并发的一种实现

 

近来研究定时器Quartz集群的实现原理时，发现了一种利用数据库锁的方式来实现集群环境下进行并发控制的方式。由于现在的系统多是部署在集群环境中，需要进行并发控制时，这是一种很好的实现方法，现将原理介绍如下：

 

首先，在数据库中建立一张拥有锁标识的表，建立表的SQL语句如下：

        CREATE TABLE TB_LOCKS

   (	LOCK_NAME VARCHAR2(40) NOT NULL, 
	PRIMARY KEY (LOCK_NAME）
   )

        表创建好之后，插入一些数据， 这些数据是根据数据库的业务逻辑操作抽象出的系统所拥有的表的类型，如"TRIGGER_ACCESS"就表示对任务触发器相关的信息进行修改删除操作时所需要获得的锁。每当要进行与某种业务相关的数据库操作时，先去这张表中查询操作相关的业务对象所需要的锁，如Quartz中需要修改触发气的状态，下次触发时间时，就需要获得"TRIGGER_ACCESS"所表示的锁。这时，执行查询这个表数据的SQL形如“select * from TB_LOCKS t where t.lock_name='TRIGGER_ACCESS' for update”，在select之后加了“for update”，就如JAVA语言中的为方法加上Symchriozed一样，起到了串形化访问修改所需数据的作用。当一个线程使用上述的SQL对表中的数据执行查询操作时，若查询结果中包含相关的行，数据库就会对这些行进行ROW LOCK。若此时又有另外一个线程使用相同的SQL对表的数据进行查询时，由于查询出的数据行已经被数据库锁住了，此时这个线程就只能等待，直到拥有这些行锁的线程完成了相关的业务操作，执行了commit动作后，数据库才会释放了相关行的锁，这个线程才能继续执行。通过这样的机制，在集群环境下，结合乐观锁的机制就可以防止一个线程对数据库数据的操作的结果被另外一个线程所覆盖，从而可以避免一些难以觉察的错误发生。当然，达到这种效果的前提是需要把Connection设置为手动提交，即autoCommit为false,下面是执行相关步骤的程序代码：

 

/**
     * Execute the given callback having optionally aquired the given lock.
     * This uses the non-managed transaction connection.
     * 
     * @param lockName The name of the lock to aquire, for example 
     * "TRIGGER_ACCESS".  If null, then no lock is aquired, but the
     * lockCallback is still executed in a non-managed transaction. 
     */    
protected Object executeInNonManagedTXLock(
            String lockName, 
            TransactionCallback txCallback) throws JobPersistenceException {
        boolean transOwner = false;
        Connection conn = null;
        try {
            if (lockName != null) {
                if (getLockHandler().requiresConnection()) {
                    conn = getNonManagedTXConnection();
                }
                //获得相关的锁，通过带有"for update"的select语句实现，如果这个锁被其他线程占用，执行这个操作的线程只能等待
                transOwner = getLockHandler().obtainLock(conn, lockName);
            }
            
            if (conn == null) {
                conn = getNonManagedTXConnection();
            }
            
            Object result = txCallback.execute(conn)；//执行相关的业务逻辑操作，获得操作的结果
            commitConnection(conn);//提交操作结果，释放锁
            return result;
        } catch (JobPersistenceException e) {
            rollbackConnection(conn);//撤销操作结果，释放锁
            throw e;
        } catch (RuntimeException e) {
            rollbackConnection(conn);//撤销操作结果，释放锁
            throw new JobPersistenceException("Unexpected runtime exception: "
                    + e.getMessage(), e);
        } finally {
            try {
                releaseLock(conn, lockName, transOwner);
            } finally {
                cleanupConnection(conn);
            }
        }
    }