分布式锁—-数据库和redis实现分布式锁

前言：

在博客“zookeeper实现分布式锁的两种方式”中介绍了分布式锁的使用场景，以及如何用zookeeper分别实现简单和高性能的分布式锁，这里就不再重复介绍分布式锁的场景，今天主要给大家带来另外两种实现分布式锁的方式–数据库、redis

一、分布式锁实现的原理：

实现分布式锁的原理基本上就是相似的，使用第三方工具做到一个互斥（排它）的作用，比如：

1、zookeeper：当客户端向zk写入节点时，如果写入成功，其他的客户端就无法写入成功，可以理解为互斥

2、数据库：向数据库插入一条数据（比如用id主键，或者唯一索引）等达到其他的客户端无法再插入相同的数据

3、redis：当一个客户端向缓存中写成功一个key-value时，其他的客户端不能在写入相同的key

知道上面的原理，实现起来就很简单了。解锁就是分别删除他们创建的节点或者数据，其他的客户端就能重新创建该节点或者数据

二、使用mysql实现分布式锁

由于mysql实现分布式锁的性能非常非常差，根本不能在线上环境使用（如果你不怕被研发经理打死可以试一下），这里就详细的说一下mysql实现的思路，具体就不用代码实现

（1）新建一张表lock

该表可以只有一个字段id，当然是主键咯，保证唯一性

（2）加锁

加锁就是在java代码中向上面的数据库中插入一条数据insert into lock （id）values （1）

如果插入成功则表示获取到锁，否则就是获取锁失败，因为就一条sql，所以这也是原子性的（加锁和解锁必须保证原子性）

（3）解锁

解锁就是删除刚才插入的数据delete from lock where id = 1

就这么简单就能使用mysql实现分布式锁，大家可以试一下

三、redis实现分布式锁

这是本文的重点，所以会详细介绍，并且会用代码实现。在实现之前，我们先考虑一下在实现的过程中应当要注意什么

（1）加锁和解锁必须保证原子性

（2）谁加的锁，应当谁解锁，不能解别人的锁

（3）当发生死锁，或者某个客户端持有的锁一直不是放怎么办？锁过期

带着这三个问题，我们再来思考下如何具体实现

1、加锁：

（1）向缓存中写入一条数据，如果该key存在则写入失败，否则写入成功，

（2）另外要设置锁的过期时间，防止一直持有锁

（3）保证（1）和（2）必须是原子性，否则有可能失效了别人的锁

jedis.set(final String key, final String value, “NX”, “PX”,final long time)这个方法就能同时满足上面的条件，NX：表示存在该key则插入失败，否则插入成功，PX：表示过期，time：表示设置的过期时间，可根据自己的业务设置，key和value就不解释了

（4）加锁的代码实现

public void lock() {
	if (tryLock()){//如果获取锁成功的话，就直接返回h
		System.out.println("线程"+Thread.currentThread().getName()+"获取到锁");
		return;
	}
	try {
		System.out.println("线程"+Thread.currentThread().getName()+"未获得到锁，进行等待...");
		Thread.sleep(10);//没获取到锁，就睡眠，有点影响性能
	} catch (InterruptedException e) {
		e.printStackTrace();
	}
	lock();//递归调用
}


@Override
public boolean tryLock() {
	//用于标记所的名字，方便解锁时判断
	String lockName = UUID.randomUUID().toString();
	//如果不存在key就插入，否则插入失败，过期时间为1s
	String ret = jedisUtils.set(KEY, lockName, "NX", "PX", 1000);
	if ("OK".equals(ret)){
		threadLocal.set(lockName);
		return true;
	}
	return false;
}

上面的加锁就实现完成了，jedisUtils.set（）是我封装了一个工具类，其实里面就是调用了jedis.set()只是封装了获取jedis的过程，另外使用了Threadlocal变量来维护锁的名字，方便解锁时线程能够获取它加锁的名字（threadlocal相当于一个副本，可跨方法，不熟悉的可以了解一下）

2、解锁

（1）解锁过程要是原子性

（2）谁加的锁，谁去解锁，不能解除它人的锁

先带大家看一段解锁的代码，看看下面代码是否能够同时满足上面两点

/**
* 错误的解锁方式
*/
public void unlockWrong(){
	String lockName = jedisUtils.get(KEY);
	//谁加锁了，就是谁解锁，防止解除他人的锁，但是这种方法不是原子性的，有问题
	if (null != lockName && lockName.equals(threadLocal.get())){
		//如果这个地方锁过期了，然后其他人抢到锁，那么它就是解除掉了他人的锁
		jedisUtils.del(KEY);
	}
}

上面的错误的解锁方式乍一看好像没什么问题，但是仔细看的话，会发现可能会解除掉别人的锁，上面注释已经写的很清楚了。那如何解决呢？要知道如何解决就需要知道上面产生错误的具体原因是什么—解锁的过程不是原子性的，但是并没有向加锁的方式一样给我们提供一个解锁的原子性的方法啊，莫慌，我们可以使用lua脚本—-lua脚本是原子性的

（3）解锁的lua脚本

if redis.call("get",KEYS[1]) == ARGV[1] then 
    return redis.call("del",KEYS[1]) 
else 
    return 0 
end

KEYS[1]：传入的第一个参数

ARGV[1]：传入的第二个参数

上面的意思就是获取某个key对应的value是否和传入的第二个参数相等，如果是则删除，其实就是上面错误解锁方法的内容

（4）解锁的代码

/**
*正确的解锁方式，应当保证原子性，应该使用lua脚本 
 */
@Override
public void unlock() {
	String script = getLuaString("unlock.lua");
	jedisUtils.eval(script, Arrays.asList(KEY),Arrays.asList(threadLocal.get()));
	System.out.println("线程"+Thread.currentThread().getName()+"释放锁成功");
}

上面就是解锁的代码，哦对了，getLuaString()是通过流的方式获取lua脚本的内容，给大家看下代码

private String getLuaString(String luaName){
	String luaPath = this.getClass().getClassLoader().getResource(luaName).getPath();
	//System.out.println("luaPath="+luaPath);
	String ret="";
	try {
		BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(new File(luaPath))));
		String temp;
		while((temp = reader.readLine()) !=null){
			ret+=temp;
		}
			
	} catch (FileNotFoundException e) {
		e.printStackTrace();
	} catch (IOException e) {
		e.printStackTrace();
	}
	return ret;
}

好了，大概就是上面这些内容了，最后我们来测试一下吧

RedisLockTest.java

package com.taolong;

import java.util.concurrent.CountDownLatch;

import com.taolong.lock.RedisLock;

public class RedisLockTest {

	private static CountDownLatch countDownLatch = new CountDownLatch(10);
	public static void main(String[] args) {
		for (int i=0;i<10;i++){
			new Thread(new LockRunnable()).start();
			countDownLatch.countDown();
		}
	}
	
	private static class LockRunnable implements Runnable{

		@Override
		public void run() {
			RedisLock redisLock = new RedisLock();
			try {
				countDownLatch.await();
				redisLock.lock();
				//模拟操作任务
				Thread.sleep(10);
			} catch (InterruptedException e) {
				e.printStackTrace();
			} finally {
				redisLock.unlock();
			}
		}
		
	}
}

运行的结果：

分布式锁----数据库和redis实现分布式锁