分布式锁实现（redis/zookeeper）

先理了解本地锁Lcok：
http://572327713.iteye.com/blog/2407789

1.基于数据库实现分布式锁
性能较差，容易出现单点故障
锁没有失效时间，容易死锁
非阻塞式的
不可重入

2.基于缓存实现分布式锁
性能好
锁失效时间难设置，容易死锁
非阻塞式的（使用线程等待解决）
不可重入

3.基于zookeeper实现分布式锁
实现相对简单
可靠性高
性能较好
可重入

数据库：
性能较差，容易出现单点故障：
mysql并发性能瓶颈300-700，一个线程访问时插入一条数据，处理后删除此数据
很容易出现单点故障，连接有瓶劲性能差
锁没有失效时间，容易死锁：
一个线程突然宕机，因为数据没有删除，其他线程一直等待--死锁
非阻塞式的：
拿锁失败后立刻返回结果，线程做其他事情
不可重入：
线程加锁，其他线程不能加锁了

redis：
性能好：
轻轻松松响应并发10万
锁失效时间难设置，容易死锁：
非阻塞式的（使用线程等待解决）：
不可重入：
线程加锁，其他线程不能加锁了
加解锁正确的姿势：（来自redis作者antirez的总结归纳）
1.加锁：
1.1生成唯一的随机值(uuid，时间搓)，向1.2 redis写入随机值完成加锁并1.3设置失效时间。
必须使用sentnx？ SET if Not exists(如果不存在，则SET)
SET resource_name my_random_value NX PX 30000
2.解锁：
2.1根据生成随机值，2.2进行比对（线程与redis里如果一致），2.3删除redis上的数据：
执行如下lua脚本
if redis.call("get".KEYS[1]) == ARGV[1] then
  return redis.call("del",KEYS[1]);
else
return 0;
end
注意！：
1.分布式锁必须要设置一个过期时间（好比unlock一定放在finally一个道理）
2.设置一个随机字符串my_random_value是很有必要的
3.加锁和设置失效时间必须是原子操作
4.释放锁保证三步原子性（get，比较del值，del）可用基于lua脚本实现
原因：a线程解锁比较redis值相等时恰好到了过期时间，此时redis写入了b线程的值，a线程会继续删除redis，导致b线程失效，所以必须保证释放锁三步原子性，使用lua脚本，不会受到b线程任何影响


图中2：a线程特意暂停3秒，但是已经超了超时时间。
图中3：a线程过了超时时间，redis清空，b线程加入了锁。
图中4：当a线程解锁时，发现此redis是b线程的时间戳，默默的离去，这就是设置随机时间搓的原因。

图中第二步： 比较redis值与本地值是否一致，删除redis进行解锁
此时本地值用的是threadlocal修饰变量。

redis的lock具体实现：
pom.xml

    <!-- https://mvnrepository.com/artifact/redis.clients/jedis -->
     <dependency>
         <groupId>redis.clients</groupId>
         <artifactId>jedis</artifactId>
         <version>2.9.0</version>
     </dependency>

RedisLock.java

package com.hailong.yu.dongnaoxuexi.lock;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

import redis.clients.jedis.Jedis;

public class RedisLock implements Lock{

	private static final String LOCK_KEY = "lock";
	
	// 线程上下文，在这个线程执行过程中，保存的变量放这里，变量传递
	private ThreadLocal<String> local = new ThreadLocal<String>();


    /**
     * 阻塞锁（synchonied是阻塞的）
     */
    public void lock() {
    	if(tryLock()) {
    		
    	} else {
    		//reids做阻塞不太灵活，用现成阻塞
    		try {
				Thread.sleep(200);
				
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
    		lock();
    	}
	}

    /**
	 * 非阻塞式锁
     */
    public boolean tryLock() {
    	
    	String uuid = UUID.randomUUID().toString();

    	Jedis redis = new Jedis("localhost");
    	// key
    	// value
    	// @param nxxx NX|XX, NX -- Only set the key if it does not already exist. XX -- Only set the key
    	// 		if it already exist. 一定要没有值才设置成功/一定要有值才能设置成功
    	// expx EX|PX, expire time units: EX = seconds; PX = milliseconds
    	// 100ms有效期
    	String ret = redis.set(LOCK_KEY, uuid, "NX", "PX", 100);
    	if (ret !=null && ret.equals("OK")) {
        	local.set(uuid);
			return true;
		}
		return false;
	}

    /**
     * 解锁
     */
    public void unlock() {
    	// FileUtils.readFileByLines("E:/workspaces/.../unlock.lua");
    	String script = "";
    	// 执行脚本命令
    	Jedis redis = new Jedis("localhost");
    	List<String> keys = new ArrayList<String>();
    	keys.add(LOCK_KEY);
    	List<String> locals = new ArrayList<String>();
    	locals.add(local.get());
    	redis.eval(script, keys, locals);
	}

	public void lockInterruptibly() throws InterruptedException {
		// TODO Auto-generated method stub
		
	}

	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
		// TODO Auto-generated method stub
		return false;
	}

	public Condition newCondition() {
		// TODO Auto-generated method stub
		return null;
	}
}

unlock.lua

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

基于redis分布式锁方案问题：
1.锁失效时间难把握，一般为单线程处理时长两到三倍
超时时间不能长也不能短，为什么设置了100ms，一般为单线程处理线程的两到三倍。
2.可能出现锁失效情况
a线程如果超时，a线程还一直以为拿着锁，出现了共享资源竞争的问题。
3.此分布式锁不能在redis集群中使用，集群环境中可用redLock。
用于单节点redis，如果再集群下不太合适，可以使用redLock复杂很多。
所以建议使用zookeeper的分布式锁。
思考：有人会说redis单点可以有主从啊，但是主从会有问题：
比如服务1获取主机锁，主机宕机，这是还没来得及同步到从机，此时丛机变主机，这时服务2获取新主机锁，这样会有两个服务获取到锁。
https://blog.csdn.net/hh1sdfsf56456/article/details/79474434


zookeeper：
1.基于内存
2.实现简单

liux
持久节点create /temp(路径) temp(值)
临时节点 create -e /temp(路径) temp(值)
顺序节点 create -s
临时顺序 create -s -e

zk应用场景：
数据发布订阅（配置中心）
命名服务
Master选举
集群管理
分布式队列
分布式锁

羊群效应在分布式集群规模比较大的环境中危害是严重的：
1.巨大的服务器性能损耗：我去发事件，我要序列化的事件啊
客户端无端接受了很多与自己无关的通知事件。
2.网络冲击，每个节点网络发事件，网络带宽消耗很大
3.当羊群效应频繁的发生，整个节点都挂了，节点可能造成宕机
如果以后集群环境2个以上节点时。

集群节点有10个，只有1个会抢占锁
存在死锁的可能性。
生产的订单号服务宕机

临时顺序节点：

package com.baozun.util.locks;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.SerializableSerializer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class ZookeeperLock implements Lock{

	private static String LOCK_PATH = "/LOCK";
	
//	@Value("${dubbo.registry.address}")
//	private static String ZK_IP_PORT;
	private static String ZK_IP_PORT = "123.206.*.*:2181";
	
	private static final Log logger = LogFactory.getLog(ZookeeperLock.class);
	
//	private ZkClient client = new ZkClient(ZK_IP_PORT, 1000, 10, new SerializableSerializer());
	private ZkClient client = new ZkClient(ZK_IP_PORT);

	private CountDownLatch cdl;

	// 之前节点
	private String beforePath;

	// 当前请求的节点
	private String currentPath;

	public ZookeeperLock() {
		if(!client.exists(LOCK_PATH)){
			client.createPersistent(LOCK_PATH);
		}
	}

    /**
     * 非阻塞式锁
     */
	@Override
	public boolean tryLock() {
		
		// 如果currentPath为空则为第一次尝试加锁，第一次加锁赋值currentPath
		if(currentPath == null || currentPath.length() <=0) {
			// 创建临时顺序节点
			currentPath = client.createEphemeralSequential(LOCK_PATH + '/', "lock");
		}
		// 获取所有临时节点并排序，临时节点名称为自增长的字符串：0000000400
		List<String> childrens = client.getChildren(LOCK_PATH);
		Collections.sort(childrens);
		// 如果当前节点在所有节点中排名第一则获取锁成功
		if(currentPath.equals(LOCK_PATH + '/' + childrens.get(0))) {
			return true;
		// 如果当前节点在所有节点中排名中不是第一，则获取前面的节点名称，并赋值给beforePath
		} else {
			int wz = Collections.binarySearch(childrens, currentPath.substring(6));
			beforePath = LOCK_PATH + '/' + childrens.get(wz-1);
		}
		return false;
	}

	@Override
	public void unlock() {
		// TODO Auto-generated method stub
		client.delete(currentPath);
	}

    /**
     * 阻塞式锁
     */
	@Override
	public void lock() {

		if(tryLock()) {
			waitForLock();
			lock();
		} else {
			logger.info(Thread.currentThread().getName()+" 获得分布式锁！");
		}
	}

	/**
	 * 等待锁
	 * @return
	 */
	public void waitForLock() {
		
		// 监听器
		IZkDataListener iZkDataListener = new IZkDataListener() {
			
			@Override
			public void handleDataDeleted(String arg0) throws Exception {

				// 节点数据被删除
				if(cdl!=null) {
					cdl.countDown();
				}
			}
			
			@Override
			public void handleDataChange(String arg0, Object arg1) throws Exception {
				// TODO Auto-generated method stub
				
			}
		};
		// 给排在前面的节点增加数据删除的wetcher
		client.subscribeDataChanges(beforePath, iZkDataListener);
		if(client.exists(beforePath)) {
			cdl = new CountDownLatch(1);
			try {
				cdl.await();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		client.unsubscribeDataChanges(beforePath, iZkDataListener);
	}

	// =========================暂不实现===========================
	@Override
	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
		// TODO Auto-generated method stub
		return false;
	}

    /** 
     * 中断机制(可中断锁) 
     */ 
	@Override
	public void lockInterruptibly() throws InterruptedException {
		// TODO Auto-generated method stub
		
	}
	
    /** 
    * 设置条件加锁或解锁
    * 多个条件变量
    */ 
	@Override
	public Condition newCondition() {
		// TODO Auto-generated method stub
		return null;
	}
}

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.ReentrantLock;

import com.baozun.util.locks.ZookeeperLock;


/**
 * @author hailong.yu1
 * @date 2018年1月25日 上午10:36:57
 */
public class SecurityProcessorTest implements Runnable {
	
	public static final int NUM = 10;
	public static CountDownLatch countDownLatch = new CountDownLatch(NUM);
	public static OrderCodeGenerator orderCodeGenerator = new OrderCodeGenerator();
	public ZookeeperLock lock = new ZookeeperLock();
	
	@Override
	public void run() {
		try {
			countDownLatch.await();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		createOrder();
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		
		for(int i=0; i<NUM; i++) {
			
			new Thread(new SecurityProcessorTest()).start();
			countDownLatch.countDown();
		}
	}
	
	public void createOrder() {
		String orderNum = null;
		
		try {
			lock.lock();
			orderNum = orderCodeGenerator.getOrderCode();
		} catch (Exception e) {
			// TODO: handle exception
		} finally {
			lock.unlock();
		}

		System.out.println(Thread.currentThread().getName()+"===="+orderNum);
	}
}

ps aux|grep java
zkServer.sh start

linux服务器中访问zk服务：


在zk服务查看znode节点：



如果线程抢占一个资源，也可以使用队列抢占解决。