Zookeeper（四）：利用ZK节点特性做共享锁功能

背景

N台服务器对同一个文件进行修改，这个时候要给这个文件加锁。
使用zk的结点名字唯一而且临时结点会话结束会清理的特性实现锁。
原理流程

Lock

package com.rzp.service;

import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

/**
 * 实现自定义的分布式锁机制
 * 
 * @author gerry
 *
 */
public class DistributedLock implements Lock, Closeable, Watcher {
    public static final Logger logger = Logger.getLogger(DistributedLock.class);
    private ZooKeeper client = null; // zk的客户端
    private String rootPath = "/distributed_lock"; // 锁机制的根节点
    private String nodePath = null; // 节点路径
    private int sleepTime = 1000;
    private String currentIp = null;

    /**
     * 传入一个lock节点名称
     * 
     * @param lockNodeName
     * @throws InterruptedException
     * @throws KeeperException
     * @throws IOException
     */
    public DistributedLock(String lockNodeName) throws IOException, KeeperException, InterruptedException {
        this("192.168.98.129:2181", lockNodeName);
    }

    /**
     * 根据传入的zk连接字符串和lock的节点名称进行初始化操作
     * 
     * @param url
     * @param lockNodeName
     * @throws InterruptedException
     * @throws KeeperException
     * @throws IOException
     */
    public DistributedLock(String url, String lockNodeName) throws IOException, KeeperException, InterruptedException {
        this(url, 48000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                // nothings
            }
        }, lockNodeName);
    }

    /**
     * 根据传入的参数构建zk客户端和lock实例
     * 
     * @param url
     * @param sessiontTimeOut
     * @param watcher
     * @param lockNodeName
     *            要求是一个只包含数字字母和下划线的单词
     * @throws IOException
     * @throws InterruptedException
     * @throws KeeperException
     */
    public DistributedLock(String url, int sessiontTimeOut, Watcher watcher, String lockNodeName)
            throws IOException, KeeperException, InterruptedException {
        // TODO: 对lockNodeName进行验证
        this.nodePath = this.rootPath + "/" + lockNodeName;
        this.client = new ZooKeeper(url, sessiontTimeOut, watcher);
        try {
            this.currentIp = java.net.InetAddress.getLocalHost().getHostName();
        } catch (Exception e) {
            this.currentIp = "unkown";
        }
        // 在jvm中添加一个client关闭的钩子
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    DistributedLock.this.close();
                } catch (IOException e) {
                    // nothings
                }
            }
        }));

        // 进行根节点是否存在的判断操作
        Stat stat = this.client.exists(this.rootPath, false);
        if (stat == null) {
            // 表示根节点不存在啊，创建
            try {
                this.client.create(this.rootPath, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            } catch (Exception e) {
                // 1. 其他的进程创建了；2. 确实是创建失败
                stat = this.client.exists(this.rootPath, false);
                if (stat == null) {
                    throw new IOException(e);
                }
            }
        }
    }

    @Override
    public void lock() {
        boolean result = false;
        while (!result) {
            try {
                result = this.internalLock();
                if (result) {
                    break; // 结束循环
                }
            } catch (Exception e) {
                logger.warn("获取锁发生异常", e);
            }

            try {
                Thread.sleep(this.sleepTime);
            } catch (InterruptedException e) {
            }
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        boolean result = false;
        while (!result) {
            result = this.internalLock();
            if (result) {
                break; // 结束循环
            }

            Thread.sleep(this.sleepTime); // 休眠1秒
        }
    }

    @Override
    public boolean tryLock() {
        boolean result = false;
        try {
            result = this.internalLock();
        } catch (Exception e) {
            logger.warn("获取锁发生异常", e);
        }
        return result;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return this.internalLock(time, unit);
    }

    /**
     * 内部的获取锁方法，返回true，表示获取锁成功
     * 
     * @return
     * @throws InterruptedException
     */
    private boolean internalLock() throws InterruptedException {
        while (true) {
            try {
                Stat stat = this.client.exists(this.nodePath, false); // 判断是否存在
                if (stat == null) {
                    // 表示不存在
                    // 创建一个节点，设置为临时节点即可
                    try {
                        this.client.create(this.nodePath, this.currentIp.getBytes(), Ids.OPEN_ACL_UNSAFE,
                                CreateMode.EPHEMERAL);
                        return true; // 获得锁
                    } catch (KeeperException e) {
                        // 创建失败, 添加监视机制
                        this.client.exists(this.nodePath, this);
                    }
                } else {
                    // 存在，所有添加监视机制
                    this.client.exists(this.nodePath, this);
                }

                // 失败
                synchronized (this) {
                    this.wait(); // 等待继续创建
                }
            } catch (KeeperException e) {
                Thread.sleep(this.sleepTime);
            }
        }
    }

    /**
     * 内部的获取锁方法，返回true，表示获取锁成功
     * 
     * @return
     * @throws InterruptedException
     */
    private boolean internalLock(long time, TimeUnit unit) throws InterruptedException {
        if (unit != TimeUnit.MILLISECONDS) {
            throw new IllegalArgumentException("时间单位必须为毫秒");
        }
        long startTime = System.currentTimeMillis();
        long millisTime = time;

        while (true) {
            try {
                Stat stat = this.client.exists(this.nodePath, false); // 判断是否存在
                if (stat == null) {
                    // 表示不存在
                    // 创建一个节点，设置为临时节点即可
                    try {
                        this.client.create(this.nodePath, this.currentIp.getBytes(), Ids.OPEN_ACL_UNSAFE,
                                CreateMode.EPHEMERAL);
                        return true; // 获得锁
                    } catch (KeeperException e) {
                        // 创建失败, 添加监视机制
                        this.client.exists(this.nodePath, this);
                    }
                } else {
                    // 节点存在, 添加监视机制
                    this.client.exists(this.nodePath, this);
                }

                // 失败
                synchronized (this) {
                    this.wait(millisTime);
                }
                long endTime = System.currentTimeMillis();
                if (endTime - startTime > millisTime) {
                    // 如果过期时间超过time
                    return false;
                }
            } catch (KeeperException e) {
                Thread.sleep(this.sleepTime);
            }
        }
    }

    @Override
    public void unlock() {
        // 解锁， 直接删除节点
        try {
            if (this.client.exists(this.nodePath, false) != null) {
                this.client.delete(this.nodePath, -1);
            }
        } catch (InterruptedException | KeeperException e) {
            throw new RuntimeException("解锁失败", e);
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    @Override
    public void close() throws IOException {
        if (this.client != null) {
            try {
                this.client.close();
            } catch (InterruptedException e) {
                throw new IOException(e);
            } finally {
                this.client = null;
            }
        }
    }

    @Override
    public void process(WatchedEvent event) {
        if (EventType.NodeDeleted.equals(event.getType()) && this.nodePath.equals(event.getPath())) {
            // 表示是节点删除操作,而且是锁节点
            synchronized (this) {
                this.notifyAll(); // 通知全部
            }
        }
    }

}

测试

package com.rzp.service;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 测试我们的分布式锁能否使用
 *
 * @author gerry
 *
 */
public class DistributedLockDemo {
    public static void main(String[] args) throws Exception {
        test3();
    }
    static void test3() throws Exception {
        final DistributedLock lock = new DistributedLock( "lock");
        int n = 10;
        final CountDownLatch latch = new CountDownLatch(1);
        final CountDownLatch latch2 = new CountDownLatch(n);
        final Random random = new Random(System.currentTimeMillis());

        for (int i = 0; i < n; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    int millis = random.nextInt(5000) + 1000;
                    try {
                        latch.await();// 等待
                        System.out.println(Thread.currentThread().getName() + "线程休眠:" + millis);
                        Thread.sleep(millis);
                    } catch (InterruptedException e) {
                    }

                    // 开始执行
                    System.out.println(Thread.currentThread().getName() + "休眠结束，开始获取锁");
                    lock.lock();
                    try {
                        System.out.println(Thread.currentThread().getName() + "获得锁 " + System.currentTimeMillis());
                        try {
                            Thread.sleep(millis);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    } finally {
                        System.out.println(Thread.currentThread().getName() + "准备释放锁");
                        lock.unlock();
                        latch2.countDown();
                    }
                }
            }, "2thread(" + i + ")").start();
        }

        latch.countDown(); // 启动
        latch2.await(); // 等待执行完成
        System.out.println("所有线程执行完");
        lock.close();
    }


}

测试结果：

每个锁都是释放后才能获取

Zookeeper（四）：利用ZK节点特性做共享锁功能

背景

猜你喜欢