《自己动手写框架4》：分布式锁的简单实现

分布式锁在分布式应用其中是要经经常使用到的，主要是解决分布式资源訪问冲突的问题。  一開始考虑採用ReentrantLock来实现，可是实际上去实现的时候。是有问题的，ReentrantLock的lock和unlock要求必须是在同一线程进行，而分布式应用中，lock和unlock是两次不相关的请求。因此肯定不是同一线程，因此导致无法使用ReentrantLock。

接下来就考虑採用自己做个状态来进行锁状态的记录，结果发现总是死锁。细致一看代码，能不锁死么。

public synchronized void lock(){
    while(lock){
        Thread.sleep(1);
    }
    lock=true;
...
}

public synchronized void unlock(){
    lock=false;
...
}

第一个请求要求获得锁。好么。给他个锁定状态，然后他拿着锁去干活了。
这个时候，第二个请求也要求锁，OK,他在lock中等待解锁。
第一个干完活了，过来还锁了，这个时候悲催了，由于。他进不了unlock方法了。

可能有人会问，为什么採用while。而不是採用wait...notify?这个问题留一下。看看有人能给出来不？
总之。上面的方安案流产了。

相同。不把synchronized 放在方法上，直接放在方法里放个同步对象能够不？？道理是一样的。也会发生上面一样的死锁。
到此为止前途一片黑暗。
@沈学良 同学的http://my.oschina.net/shenxueliang/blog/135865写了一个用zk做的同布锁，感觉还是比較复杂的且存疑。

自己做不出来吧。又不死心。

再来看看Lock的接口，想了一下，不遵守Lock的接口了。编写了以下的接口。

public interface DistributedLock extends RemoteObject {

    long lock() throws RemoteException, TimeoutException;

    long tryLock(long time, TimeUnit unit) throws RemoteException, TimeoutException;

    void unlock(long token) throws RemoteException;

}

呵呵，眼尖的同学可能已经发现不同了。

lock方法添加了个long返回值。tryLock方法，返回的也不是boolean。也是long，unlock方法多了一个long參数型參数，呵呵。技巧就在这里了。

public class DistributedLockImpl extends UnicastRemoteObject implements DistributedLock {
    /**
     * 超时单位
     */
    private TimeUnit lockTimeoutUnit = TimeUnit.SECONDS;
    /**
     * 锁的令牌
     */
    private volatile long token = 0;
    /**
     * 同步对象
     */
    byte[] lock = new byte[0];
    /**
     * 默认永不超时
     */
    long lockTimeout = 60 * 60;//默认超时3600秒
    long beginLockTime;//获取令牌时间。单位毫秒


    public DistributedLockImpl() throws RemoteException {
        super();
    }


    /**
     * @param lockTimeout 锁超时时间。假设加锁的对象不解锁，超时之后自己主动解锁
     * @param lockTimeoutUnit 
     * @throws RemoteException
     */
    public DistributedLockImpl(long lockTimeout, TimeUnit lockTimeoutUnit) throws RemoteException {
        super();
        this.lockTimeout = lockTimeout;
        this.lockTimeoutUnit = this.lockTimeoutUnit;
    }
    public long lock() throws TimeoutException {
        return tryLock(0, TimeUnit.MILLISECONDS);
    }
    private boolean isLockTimeout() {
        if (lockTimeout <= 0) {
            return false;
        }
        return (System.currentTimeMillis() - beginLockTime) < lockTimeoutUnit.toMillis(lockTimeout);
    }
    private long getToken() {
        beginLockTime = System.currentTimeMillis();
        token = System.nanoTime();
        return token;
    }
    public long tryLock(long time, TimeUnit unit) throws TimeoutException {
        synchronized (lock) {
            long startTime = System.nanoTime();
            while (token != 0 && isLockTimeout()) {
                try {
                    if (time > 0) {
                        long endTime = System.nanoTime();
                        if (endTime - startTime >= unit.toMillis(time)) {
                            throw new TimeoutException();
                        }
                    }
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    //DO Noting
                }
            }
            return getToken();
        }
    }
    public void unlock(long token) {
        if (this.token != 0 && token == this.token) {
            this.token = 0;
        } else {
            throw new RuntimeException("令牌" + token + "无效.");
        }
    }
}

以下对代码进行一下解说。

上面的代码提供了，永远等待的获取锁的lock方法和假设在指定的时间获取锁失败就获得超时异常的tryLock方法，另外另一个unlock方法。
技术的关键点实际上就是在token上，上面的实现。有一个主要的假设，就是两次远程调用之间的时间不可能在1纳秒之内完毕。

因此，每次锁的操作都会返回一个长整型的令牌，就是当时执行时间的纳秒数。下次解锁必须用获得的令牌进行解锁，才干够成功。如此，解锁就不用加入同步操作了，从而解决掉上面死锁的问题。
实际上，没有令牌也是能够的，可是那样就会导致a获取了锁，可是b执行unlock也会成功解锁。是不安全的。而加入令牌。就能够保证仅仅有加锁者才干够解锁。
以下是測试代码：

public class TestDLock {
    public static void main(String[] args) throws Exception {
        RmiServer rmiServer = new LocalRmiServer();
        DistributedLockImpl distributedLock = new DistributedLockImpl();
        rmiServer.registerRemoteObject("lock1", distributedLock);
        MultiThreadProcessor processor = new MultiThreadProcessor("aa");
        for (int i = 0; i < 8; i++) {
            processor.addProcessor(new RunLock("aa" + i));
        }
        long s = System.currentTimeMillis();
        processor.start();
        long e = System.currentTimeMillis();
        System.out.println(e - s);
        rmiServer.unexportObject(distributedLock);
    }
}

class RunLock extends AbstractProcessor {
    public RunLock(String name) {
        super(name);
    }

    @Override
    protected void action() throws Exception {
        try {
            RmiServer client = new RemoteRmiServer();
            DistributedLock lock = client.getRemoteObject("lock1");
            for (int i = 0; i < 1000; i++) {
                long token = lock.lock();
                lock.unlock(token);
            }
            System.out.println("end-" + Thread.currentThread().getId());
        } catch (RemoteException e) {
            e.printStackTrace();
        }
    }
}

执行情况：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27

-0    [main] INFO   - 线程组<aa>执行開始,线程数8... -3    [aa-aa0] INFO   - 线程<aa-aa0>执行開始... -3    [aa-aa1] INFO   - 线程<aa-aa1>执行開始... -3    [aa-aa2] INFO   - 线程<aa-aa2>执行開始... -3    [aa-aa3] INFO   - 线程<aa-aa3>执行開始... -3    [aa-aa4] INFO   - 线程<aa-aa4>执行開始... -4    [aa-aa5] INFO   - 线程<aa-aa5>执行開始... -4    [aa-aa6] INFO   - 线程<aa-aa6>执行開始... -8    [aa-aa7] INFO   - 线程<aa-aa7>执行開始... end-19 -9050 [aa-aa3] INFO   - 线程<aa-aa3>执行结束 end-17 -9052 [aa-aa1] INFO   - 线程<aa-aa1>执行结束 end-20 -9056 [aa-aa4] INFO   - 线程<aa-aa4>执行结束 end-16 -9058 [aa-aa0] INFO   - 线程<aa-aa0>执行结束 end-21 -9059 [aa-aa5] INFO   - 线程<aa-aa5>执行结束 end-26 -9063 [aa-aa7] INFO   - 线程<aa-aa7>执行结束 end-18 -9064 [aa-aa2] INFO   - 线程<aa-aa2>执行结束 end-22 -9065 [aa-aa6] INFO   - 线程<aa-aa6>执行结束 -9066 [main] INFO   - 线程组<aa>执行结束, 用时:9065ms 9069

也就是9069ms中执行了8000次锁定及解锁操作。
小结：
上面的分布式锁实现方案，综合考虑了实现简单，锁安全，锁超时等因素。

实际測试，大概900到1000次获取锁和释放锁操作每秒，能够满足大多数应用要求。

---

 

欢迎訪问开源技术社区：http://bbs.tinygroup.org。

本例涉及的代码和框架资料，将会在社区分享。《自己动手写框架》成员QQ群：228977971，让我们一起动手。了解开源框架的奥秘！