首先考虑一个场景，来源与实际工作中的转让场景，其正向流程大致为:发起转让--->转让校验---->记录请求表--->生成产品上架---->交易成功-----变更转让请求状态为成功，反向流程为:(产品还未交易之前)发起撤销---->变更转让请求状态为已撤销。
假如客户端同时发起了上千个请求，这些请求都是对同一笔资产发起转让，如何保证最终只有一次转让发起成功？这就需要有把锁，且只有一个请求拿到锁后，其他请求不能再拿到锁，拿到锁的请求才能发起转让。
首先，从数据表设计上uk是必须要有的，在这个前提下，我们考虑下解决方案：
方案1，根据数据库的uk限制，最终保证只有一个转让记录能够落数据表成功。但是这里问题来了，这么多的请求都往db上打去，db压力太大了，并且有效的请求只有一个，其他的请求根本就是在做无用功。我们知道db是稀有资源，在任何系统设计中都应该受到保护，所以这种方案只有在竞争度很小的时候可行。
方案2，考虑分布式锁，只有获取到锁的这个请求最终能转让成功，这样可以把其他无用的请求挡在了db之外，可行。那如何实现分布式锁呢？
市面上大概有两种方式来实现分布式锁，一是基于缓存实现，其又分redis和memcache两种，一是基于zookeeper实现。下面我们看下redis如何实现:
private static boolean lock(String key){
Jedis jedis = new Jedis("127.0.0.1", 6379);
String value = jedis.set(key,key,"NX","EX",1*60*60);
if("OK".equalsIgnoreCase(value)){
return true;
}
System.out.println("current thread "+ Thread.currentThread().getName()+ " does not get lock,key:"+key);
return false;
}
主要是借助于redis setnx指令实现，这里的key=资产id+"_transfer_apply",过期时间1小时。表示发起转让，因为撤销的时候也有可能同时多个请求对同一笔资产转让发起撤销，所以我们的key加上了业务场景。这里考虑一个问题，这个锁该什么时候释放？
分析一下上面的场景，当转让请求状态变更为成功时，这时资产已经转让成功，此时这笔资产就不能再次发起转让了(页面上没有转让按钮)，这时就可以考虑删除这个key了。另一个释放的场景是转让撤销后，撤销后资产可以再次发起转让，所以撤销的时候也要删除这个key。
首先考虑下转让成功的场景，在转让成功后，我们需要变更请求状态为成功，同时要删除key=资产id+"_transfer_apply"，这里有个问题如果我们请求状态变更成功，但是redis客户端挂了，没有删除key(如果没有删除的key是少数，那影响不大，如果此时由大量的请求，创建了大量的key，执行这步操作的时候客户端异常了，那影响大了)，如何处理？
一种方式是把redis操作和变更状态放入到同一个事务中，如果操作redis失败，回滚变更状态的操作，然后依靠job重试保证最终可以成功。不好的是与业务逻辑耦合了。
另一种方式是采用异步操作，变更状态后，发送mq消息，在消费端处理redis的操作，依靠mq和job重试保证最终key被删除了。
再考虑一种场景，在资产转让成功后，客户端构造大量并发请求进行转让，如何处理？这种情况我们也使用锁(key=资产id+"_transfer_apply")，过期时间半小时，保证只有一个请求可以走到后端，因为此时资产id请求过来后，首先要通过资产表获取资产的信息，由于已经被转让，是一个结束的状态，此时直接返回不可转让。
再考虑一个极端场景，用户写了个脚本，一天到晚对同一笔资产发起恶意的转让，这种情况下，就得引入流控机制了(另一篇文中详细说流控)。
接下来我们再来考虑下撤销的场景，撤销是一个用户行为，可能同一时间大量的并发请求进来，所以也需要一个锁，key=资产id+"_transfer_cancel"，其过期时间为1分钟，需要做的事情为1，校验是否可以取消，2，变更状态为取消，3，删除key=资产id+"_transfer_apply",这种情况下如何处理？
一种方式将redis操作和变更放入事务中，如果redis操作失败，回滚变更状态，然后考job重试最终保证，这个方案不好的是与业务逻辑耦合了。
另一种方式是变更状态后，发送mq消息，在消费端处理redis操作，依靠mq消息和job重试保证最终key被删除了
再考虑一种场景，在资产转让成功后，客户端构造大量并发请求进行撤销，如何处理？这种情况我们也使用锁(key=资产id+"_transfer_cancel")，过期时间1分钟，保证只有一个请求可以走到后端，因为此时资产id请求过来后，首先要通过请求表获取请求的信息，由于已经撤销，此时直接返回撤销成功。
再考虑一个极端场景，用户写了个脚本，一天到晚对同一笔资产发起恶意的转让，这种情况下，就得引入流控机制了(另一篇文中详细说流控)。
以上都是基于setnx实现的分布式锁，结合具体业务场景进行了了加锁以及释放锁的分析。其实redis分布式锁的实现我们还可以依靠计数器的方式实现,这里跟流控很类似，只需要把limit设置为1即可满足我们的需要，具体的流控请看"关于分布式系统流控的思考".
接下来，我们看下，基于memcache如何实现分布式锁,memcache中的add命令即可实现与redis setnx命令一样的功能，这里就不再贴代码了。
接下来，我们看下基于zookeeper的分布式锁实现,其原理是：首先创建一个锁的根节点，想要获取所的客户端在根节点下创建znode，名称类似xxx-lock-000001,xxx-lock-000002.....带有序列号的znode。然后客户端读取根节点下的所有子节点，进行排序，如果排序的第一个值根自己创建的znode一致，则认为自己获取到了锁，否则认为未获取到锁。比如客户端1创建了节点xx-lock-002，客户端2创建了节点xx-lock-001,客户端3创建了节点xx-lock-003,那此时客户端2获取到了锁。接下来我们看下代码实现，借助于curator客户端，我们可以很方便的实现：
public class ZookeeperMain {
private static final String LOCK_PATH_PREFIX = "/mylock/";
private static final String IP = "127.0.0.1";
public static void main(String[] args) throws Exception {
for(int i=0;i<10;i++){
Thread thread = new Thread("thread-"+i){
@Override
public void run() {
getLock(LOCK_PATH_PREFIX+IP);
}
};
thread.start();
}
}
private static void getLock(String lockPath){
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);//1s重试3次
final CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", retryPolicy);
client.start();
final InterProcessMutex lock = new InterProcessMutex(client, LOCK_PATH_PREFIX+IP);
try {
boolean result = lock.acquire(10,TimeUnit.MILLISECONDS);
if(result){
System.out.println(Thread.currentThread().getName()+" get the lock");
Thread.sleep(3000);//for test
}else{
System.out.println(Thread.currentThread().getName()+" did not get the lock");
}
} catch (Exception e) {
//e.printStackTrace();//comment for test
}finally{
try {
lock.release();
} catch (Exception e2) {
//e2.printStackTrace();
}
client.close();
}
}
}
最后我们分析一下这两种分布式锁的应用场景，首先看下zookeeper,因为需要创建lockpath,在并发量很大的情况下频繁创建文件，对性能可能有较大的影响，所以得根据实际业务场景使用。实际工作中，我们使用的场景是job调度，由于可以执行job的机器有多台（节点），触发调度后需要选出一个节点执行任务。
对于redis实现的分布式锁，可以支撑大并发量的访问，由于都是内存操作，所以性能较好，但有个场景就是redis主从切换时可能导致两个请求同时获得锁(从这点看，zookeeper的稳定性更好)，这时就需要看具体业务场景是否允许，如果业务有数据库写操作，可以通过uk和乐观锁解决这个问题。
综合来看，如果业务场景对稳定性，强一致性有非常高的要求，那么建议使用zookeeper锁,如果对并发度要求非常高，那可以考虑使用redis。
再思考一个问题，以上的实现是不可重入锁，假如需要实现可重入的分布式锁呢，如何实现？以redis为例子，当前线程如果获取到锁后，我们需要记录下获取锁成功的线程id,
下一次当前线程再获取的时候，需要判断当前线程id和redis中存储的线程id是否一致，如果一致，即认为获取到了锁:
public static boolean reentrantLock(String key){
Jedis jedis = new Jedis("127.0.0.1", 6379);
long threadId = Thread.currentThread().getId();
String value = jedis.set(key,String.valueOf(threadId),"NX","EX",1*60*60);
if("OK".equalsIgnoreCase(value)){
System.out.println("current thread "+ Thread.currentThread().getName()+ " get lock,key:"+key);
return true;
}else{
String val = jedis.get(key);
if(null != val){
long cachedThreadId = Long.valueOf(val);
if(cachedThreadId == threadId){
System.out.println("current thread "+ Thread.currentThread().getName()+ " get lock,key:"+key);
return true;
}
}
}
System.out.println("current thread "+ Thread.currentThread().getName()+ " does not get lock,key:"+key);
return false;
}
