分布式系统-数据一致性问题04（分布式锁）

为什么要使用分布式锁

• 为了保证一个方法在高并发情况下的同一时间只能被同一个线程执行，在传统单体应用单机部署的情况下，可以使用Java并发处理相关的API(如ReentrantLcok或synchronized)进行互斥控制。但是，随着业务发展的需要，原单体单机部署的系统被演化成分布式系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题。

分布式锁应该具备哪些条件

• 在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行；
• 高可用的获取锁与释放锁；
• 高性能的获取锁与释放锁；
• 具备可重入特性；
• 具备锁失效机制，防止死锁；
• 具备阻塞锁特性，即没有获取到锁将继续等待获取锁；
• 具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败。

基于数据库实现--记录表

• 要实现分布式锁，最简单的方式可能就是直接创建一张锁表，然后通过操作该表中的数据来实现了。
• 当我们要锁住某个方法或资源时，我们就在该表中增加一条记录，想要释放锁的时候就删除这条记录。
• 创建这样一张数据库表：

CREATE TABLE  `methodLock`  (
`id`  int(11)  NOT NULL AUTO_INCREMENT COMMENT  '主键',
`method_name`  varchar(64)  NOT NULL DEFAULT  ''  COMMENT  '锁定的方法名',
`desc`  varchar(1024)  NOT NULL DEFAULT  '备注信息',
`update_time`  timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT  '保存数据时间，自动生成',
PRIMARY KEY  (`id`),
UNIQUE KEY  `uidx_method_name`  (`method_name `)  USING BTREE
)  ENGINE=InnoDB  DEFAULT CHARSET=utf8 COMMENT='锁定中的方法';

• 当我们想要锁住某个方法时，执行以下SQL：

insert  into  methodLock(method_name,desc)  values  (‘method_name’,‘desc’)

• 因为我们对method_name做了唯一性约束，这里如果有多个请求同时提交到数据库的话，数据库会保证只有一个操作可以成功，那么我们就可以认为操作成功的那个线程获得了该方法的锁，可以执行方法体内容。
• 当方法执行完毕之后，想要释放锁的话，需要执行以下Sql：delete from methodLock where method_name ='method_name'

上面这种简单的实现有以下几个问题：

1. 这把锁强依赖数据库的可用性，数据库是一个单点，一旦数据库挂掉，会导致业务系统不可用。
2. 这把锁没有失效时间，一旦解锁操作失败，就会导致锁记录一直在数据库中，其他线程无法再获得到锁。
3. 这把锁只能是非阻塞的，因为数据的insert操作，一旦插入失败就会直接报错。没有获得锁的线程并不会进入排队队列，要想再次获得锁就要再次触发获得锁操作。
4. 这把锁是非重入的，同一个线程在没有释放锁之前无法再次获得该锁。因为数据中数据已经存在了。

当然，我们也可以有其他方式解决上面的问题。

1. 数据库是单点？搞两个数据库，数据之前双向同步。一旦挂掉快速切换到备库上。
2. 没有失效时间？只要做一个定时任务，每隔一定时间把数据库中的超时数据清理一遍。
3. 非阻塞的？搞一个while循环，直到insert成功再返回成功。
4. 非重入的？在数据库表中加个字段，记录当前获得锁的机器的主机信息和线程信息，那么下次再获取锁的时候先查询数据库，如果当前机器的主机信息和线程信息在数据库可以查到的话，直接把锁分配给他就可以了。

基于数据库实现--排他锁

除了可以通过增删操作数据表中的记录以外，其实还可以借助数据中自带的锁来实现分布式的锁。

我们还用刚刚创建的那张数据库表。可以通过数据库的排他锁来实现分布式锁。 基于MySql的InnoDB引擎，可以使用以下方法来实现加锁操作：

public  boolean  lock(){
connection.setAutoCommit(false)
while(true){
  try{
     result  =  select  *  from  methodLock  where  method_name=xxx  for  update;
     if(result==null){  return  true;  }
  }catch(Exception  e){
  }
sleep(1000);
}
return  false;}

在查询语句后面增加for update，数据库会在查询过程中给数据库表增加排他锁（这里再多提一句，InnoDB引擎在加锁的时候，只有通过索引进行检索的时候才会使用行级锁，否则会使用表级锁。这里我们希望使用行级锁，就要给method_name添加索引，值得注意的是，这个索引一定要创建成唯一索引，否则会出现多个重载方法之间无法同时被访问的问题。重载方法的话建议把参数类型也加上。）。当某条记录被加上排他锁之后，其他线程无法再在该行记录上增加排他锁。

基于缓存实现--REDIS

使用Redis来实现分布式可以基于一下三个命令

1. setnx（k,v）：想redis中set一个key为val的字符串，若key已存在，则什么都不做返回0，若key不存在则设置成功，并返回1，（同时val的值设置为一个UUID，用于在释放锁的时候进行判断，防止误解锁）
2. expire(k,time)： 为上面设置号的key设置一个超时时间，超过该时间则自动释放锁（防止死锁）
3. delete（ key）当业务执行完成后，使用delete命令删除锁，进行锁的释放（在释放锁时通过第一步中返回的UUID判断是否时该锁）

TIP: 在获取锁的时候，需要设置一个超时时间，

/**
     * 加锁
     * @param locaName  锁的key
     * @param acquireTimeout  获取超时时间
     * @param timeout   锁的超时时间
     * @return 锁标识
     */
    public String acquire(String locaName, long acquireTimeout, long timeout) {
        Jedis conn = null;
        String retIdentifier = null;
        try {
            // 获取连接
            conn = jedisPool.getResource();
            // 随机生成一个value
            String identifier = UUID.randomUUID().toString();
            // 锁名，即key值
            String lockKey = "lock:" + locaName;
            // 超时时间，上锁后超过此时间则自动释放锁
            int lockExpire = (int)(timeout / 1000);

            // 获取锁的超时时间，超过这个时间则放弃获取锁
            long end = System.currentTimeMillis() + acquireTimeout;
            while (System.currentTimeMillis() < end) {
                if (conn.setnx(lockKey, identifier) == 1) {
                    conn.expire(lockKey, lockExpire);
                    // 返回value值，用于释放锁时间确认
                    retIdentifier = identifier;
                    return retIdentifier;
                }
                // 返回-1代表key没有设置超时时间，为key设置一个超时时间
                if (conn.ttl(lockKey) == -1) {
                    conn.expire(lockKey, lockExpire);
                }

                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        } catch (JedisException e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                conn.close();
            }
        }
        return retIdentifier;
    }

/**
     * 释放锁
     * @param lockName 锁的key
     * @param identifier    释放锁的标识
     * @return
     */
    public boolean release(String lockName, String identifier) {
        Jedis conn = null;
        String lockKey = "lock:" + lockName;
        boolean retFlag = false;
        try {
            conn = jedisPool.getResource();
            while (true) {
                // 监视lock，准备开始事务
                conn.watch(lockKey);
                // 通过前面返回的value值判断是不是该锁，若是该锁，则删除，释放锁
                if (identifier.equals(conn.get(lockKey))) {
                    Transaction transaction = conn.multi();
                    transaction.del(lockKey);
                    List<Object> results = transaction.exec();
                    if (results == null) {
                        continue;
                    }
                    retFlag = true;
                }
                conn.unwatch();
                break;
            }
        } catch (JedisException e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                conn.close();
            }
        }
        return retFlag;
    }

在分布式环境中，对资源进行上锁有时候是很重要的，比如抢购某一资源，这时候使用分布式锁就可以很好地控制资源，当然真正的大规模抢购，仅仅是设置分Redis布式锁时不够的，还会出现类似Redis的热点key等问题。。。