分布式锁

分布式锁产生原因

       为了保证在分布式领域*享数据安全的问题。例如：在服务器集群的环境中，整个生成时间戳代码都是相同的，如果多个服务器在相同的时间内，使用时间戳方式生成全局id，id是相同的，不能保证全局id是唯一。在同一个时间点，只能保证有一台服务器生成全局id，这样就引入了分布式锁的概念。

解决方案

• 数据库实现分布式锁（不推荐、效率特别低）
• 基于Redis实现分布式锁setNx（考虑死锁、释放问题）
• 基于Zookeeper实现分布式锁（强烈推荐）实现相对简单，使用临时节点释放锁（效率最高）、失效时间容易控制

数据库实现分布式锁

       原理很简单，多个服务器生成相同的id，看谁能在数据库中插入成功（id保证唯一），谁就能获得锁，没有成功的，就继续等待。这样的方式不推荐，效率很低，就不过多演示了。

基于Redis实现分布式锁

      原理：setNx也可以存入key，它的特殊点在于如果存入key成功，就返回1，如果已经存在该key值，就返回0。多个客户端，使用setNx命令方式，同事在redis上创建相同的key，因为redis的key不能够重复，只要谁能够创建key成功，谁就能获得锁，没有创建成功的，就会继续等待。

       这样方式有个弊端，有可能造成死锁现象。如果一个客户端创建key成功，但是在执行完操作的时候，没有删除对应的key，那其他的客户端就会一直等待下去。

       解决方案就是对每个对应的key设置自己的有效期，有效期的目的就是为了防止死锁现象。

       在基于Redis实现分布式锁之前，需要熟系Redis三个命令：

• SETNX

SETNX key val当且仅当key不存在时，set一个key为val的字符串，返回1；若key存在，则什么都不做，返回0。

• Expire

expire key timeout，为key设置一个超时时间，单位为second，超过这个时间锁会自动释放，避免死锁。

• Delete

delete key，删除key。

1、pom文件引入

<dependencies>
	<dependency>
		<groupId>redis.clients</groupId>
		<artifactId>jedis</artifactId>
		<version>2.9.0</version>
	</dependency>
</dependencies>

2、后端相关代码

public class LockRedis {
	// redis线程池
	private JedisPool jedisPool;
	// 同时在redis上创建相同的一个key 相同key 名称
	private String redislockKey = "redis_lock";

	public LockRedis(JedisPool jedisPool) {
		this.jedisPool = jedisPool;
	}
	/**
	 * 获得锁
	 * @param acquireTimeout
	 * 			  在获取锁之前的超时时间
	 *在尝试获取锁的时候，如果在规定的时间内还没有获取锁，直接放弃。
	 * @param timeOut
	 * 			  在获取锁之后的超时时间
	 *当获取锁成功之后，对应的key 有对应有效期，对应的key 在规定时间内进行失效
	 * @return
	 */
	public String getRedisLock(Long acquireTimeout, Long timeOut) {
		Jedis conn = null;
		try {
			// 1.建立redis连接
			conn = jedisPool.getResource();
			// 2.定义 redis 对应key 的value值( uuid) 作用 释放锁 随机生成value
			String identifierValue = UUID.randomUUID().toString();
			// 3.定义在获取锁之后的超时时间
			int expireLock = (int) (timeOut / 1000);// 以秒为单位
			// 4.定义在获取锁之前的超时时间
			// 5.使用循环机制 如果没有获取到锁，要在规定acquireTimeout时间 保证重复进行尝试获取锁（乐观锁）
			// 使用循环方式重试的获取锁
			Long endTime = System.currentTimeMillis() + acquireTimeout;
			while (System.currentTimeMillis() < endTime) {
				// 获取锁
				// 6.使用setnx命令插入对应的redislockKey ，如果返回为1 成功获取锁
				if (conn.setnx(redislockKey, identifierValue) == 1) {
					// 设置对应key的有效期
					conn.expire(redislockKey, expireLock);
					return identifierValue;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (conn != null) {
				conn.close();
			}
		}
		return null;
	}

	/**
	 *释放redis锁
	 *释放锁有两种 key自动有有效期
	 * 整个程序执行完毕情况下，删除对应key
	 * 如果直接使用 conn.del(redislockKey); 保证对应是自己的创建redislockKey 删除对应自己的
	 * @param identifierValue
	 */
	public void unRedisLock(String identifierValue) {
		Jedis conn = null;
		// 1.建立redis连接
		conn = jedisPool.getResource();
		try {
			// 如果该锁的id 等于identifierValue 是同一把锁情况才可以删除
			if (conn.get(redislockKey).equals(identifierValue)) {
				System.out.println("释放锁..." + Thread.currentThread().getName() + ",identifierValue:" + identifierValue);
				conn.del(redislockKey);
			}
		} catch (Exception e){
		} finally {
			if (conn != null) {
				conn.close();
			}
		}
	}
}

基于Zookeeper实现分布式锁

1、pom文件引入

<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.10</version>
</dependency>

2、后端相关代码

public class ZKLock {
    // 集群连接地址
    protected String CONNECTION = "127.0.0.1:2181";
    // zk客户端连接
    protected ZkClient zkClient = new ZkClient(CONNECTION);
    // path路径
    protected String lockPath = "/path";
    //信号量
    protected CountDownLatch countDownLatch = new CountDownLatch(1);

    //获得锁
    public void getLock() {
        //尝试获取锁
        boolean isGetLock = tryLock();
        if(isGetLock){
            System.out.println("####获取锁成功######");
        }else {
            //进入等待
            waitLock();
            //重新获取锁
            getLock();
        }
    }

    // 尝试获取锁
    private boolean tryLock(){
        try {
            zkClient.createEphemeral(lockPath);
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    //进入等待
    private void waitLock() {
        //使用zk临时事件监听
        IZkDataListener iZkDataListener = new IZkDataListener() {
            //删除节点通知
            public void handleDataDeleted(String path) throws Exception {
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
            }
            //更改节点通知
            public void handleDataChange(String arg0, Object arg1) throws Exception {

            }
        };
        // 注册事件通知
        zkClient.subscribeDataChanges(lockPath, iZkDataListener);
        if (zkClient.exists(lockPath)) {
            countDownLatch = new CountDownLatch(1);
            try {
                //只要计数器不为0，就一直等待
                countDownLatch.await();
            } catch (Exception e) {
                // TODO: handle exception
            }
        }
        // 监听完毕后，移除事件通知
        zkClient.unsubscribeDataChanges(lockPath, iZkDataListener);
    }

    //释放锁
    public void unLock() {
        if (zkClient != null) {
            System.out.println("#######释放锁#########");
            zkClient.close();
        }
    }
}

三种分布式锁的对比

• 从实现的复杂性角度（从低到高）

       Zookeeper >= 缓存 > 数据库

• 从性能角度（从高到低）

       缓存 > Zookeeper >= 数据库

• 从可靠性角度（从高到低）

      Zookeeper > 缓存 > 数据库