最近做一个统计日志的业务（log+redis），本来在单机环境正常开车的，突然上线就掉坑了...
流程：去redis中取出当日的数据 --> 各种计算后累加计算结果，并再次存入redis -->定时任务每日24点将redis数据存入日志文件中。

1.问题引出

先上代码 (单机环境)

  private synchronized void addRecord() {
        try {
          //获取redis记录
            DeliveryQueryRecord record = backRedisService.getDeliveryQueryRecord();
            if (record == null) {
                record = new DeliveryQueryRecord();
            }
            // xxx数据操作
            //将数据再次保存到redis中
            backRedisService.setDeliveryQueryRecord(record);
        } catch (Exception e) {
            e.printStackTrace();
        }  
    }

由于使用了synchronized 关键字可以保证每个线程都会同步阻塞的执行，也就是说一个线程获取redis记录并且各种操作写入redis之后其他线程才能读和写（针对addRecord方法的，不是真正意义上的读写）进而保证数据正确性。
但是这么写单机环境下ok，可是线上是分布式集群环境，当不止一台机器同时多线程执行addRecord方法就可能存在数据同步问题。例如：A机器上有一个线程1去获取redis记录，这时同时机器B也有一个线程去获取redis记录，虽然我们知道 redis是单线程同步阻塞 的，但是极端情况下，在线程1还没写入redis之前，线程2可能会去获取数据，而此时的数据和线程1获取的是一模一样的，而不是获取线程1处理之后的数据，也就是说两个线程不是同步阻塞获取那么数据就是有问题的。

2.解决方法

(多机环境)
问题：

多机环境下 synchronized关键字不能保证所有的线程是同步阻塞运行的,也就是addRecord方法不再具备原子性

解决思路：

我们利用redis模拟一个线程锁,保证在多机环境下addRecord方法任然具备原子性，那这个线程锁如何实现呢？
我们可以利用redis的特性单线程原子性做一个特殊的KEY，这个KEY在addRecord方法真正执行前，要求线程先去查询KEY是否存在。

存在条件：
A：如果KEY不存在，说明当前没有任何线程占有锁，当前线程可以获取锁（写入KEY），随后进行数据操作，业务代码完毕后一定要释放锁(最好在finally{}代码块中实现)，随后别的线程可以竞争锁。注意在这个期间为了保证原子性，不允许任何其他线程获取锁,只能等待当前线程释放锁后,其他线程才能竞争获取。
B：如果KEY存在，说明有线程正在操作数据(获取锁)，需要等待获得锁的线程操作完毕并释放锁后，再和其他等待的线程一起竞争(非公平锁)，来获取锁。
总结：线程执行addRecord方法前必须获取锁,方法执行完毕后要释放锁，保证操作的原子性。

工具筹备

工欲善其事，必先利其器。有了牛X的工具代码，业务代码也就迎刃而解。但是在展示利器之前先展示下错误的“坑”。

加锁坑：( 错误示例)

既然已经理清了redis分布式锁实现的思路，那就上代码

public boolean tryGetLock(String lockKey, String value, int expireTime) {
        Jedis redis = null;
        try {
            redis = redisManager.redisPool.getResource();
            boolean flag = redis.exists(key);//判断key是否存在
            if (!flag) {//不存在则设置key
                String result = redis.set(lockKey,expireTime, value);
                if (LOCK_SUCCESS.equals(result)) {
                    return true;
                }
                return false;
            }

        } finally {
            if (null != redis) {
                redis.close();
            }
        }
    }

此坑的问题细心的童鞋已经发现，boolean flag = redis.exists(key);//判断key是否存在与String result = redis.set(lockKey, value);是两步操作。也就意味着不是原子操作，可能会存在多个线程在没有执行redis.set(lockKey, value);之前，通过boolean flag = redis.exists(key);可能都会得到key不存在，这样多个线程都可以继续写入key，造成了多个线程同时获得锁,不能形成同步阻塞执行。

利器代码

    private static final String LOCK_SUCCESS = "OK";
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "EX"; //px:毫秒 ex秒
    private static final Long RELEASE_SUCCESS = 1L;

    @Autowired
    RedisManager redisManager;

    /**
     * 尝试获取分布式锁
     *
     * @param lockKey    锁
     * @param requestId  请求标识
     * @param expireTime 超期时间
     * @return 是否获取成功
     */
    public boolean tryGetLock(String lockKey, String requestId, int expireTime) {
        Jedis redis = null;
        try {

            redis = redisManager.redisPool.getResource();
            String result = redis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
            if (LOCK_SUCCESS.equals(result)) {
                return true;
            }
            return false;
        } finally {
            if (null != redis) {
                redis.close();
            }
        }
    }

在redisxxx.xxx版本之后，redis支持set多参数方法：set(KEY,VALUE,SET_IF_NOT_EXIST,SET_WITH_EXPIRE_TIME,EXPIRE_TIME)
KEY,VALUE 不必多说
SET_IF_NOT_EXIST 处理模式：NX表示如果KEY不存在则插入成功，否则不插入数据此处正好是把 两步操作合并,1查询key是否存在 2修改数据,这样由于redis的单线程阻塞，保证了两步操作的原子性，即阻塞执行
SET_WITH_EXPIRE_TIME 过期时间单位: px:毫秒 ex秒
EXPIRE_TIME 过期时间：保证在宕机等其他原因不能正常删除KEY（释放锁）时,利用过期删除，从而避免KEY没有删除而造成其他线程永久等待的死锁。

此方法完美解决了上锁坑的问题。

放锁坑：（错误示例）

     public void unLock(String lockKey) {
       Jedis redis = null;
       try {
           redis = redisManager.redisPool.getResource();
           redis.del(lockKey);
       } finally {
           if (null != redis) {
               redis.close();
           }
       }
   }

对于redis分布式锁简单放锁，直接删除。很明显要是能这么简单，就不会有这一节了。那么这么操作存在什么问题呢？其实这个问题是比较隐蔽的。想象一个极端的例子：在线程1执行代码时耗时比较久，甚至超过了KEY的有效期。此时线程1还没有释放锁，线程2发现KEY没了，可以获取锁了，于是线程2写入KEY获取了锁，这时线程1执行完毕，开始释放锁对KEY进行删除。但问题是KEY锁的拥有者已经不是线程1，而是线程2。这时删除KEY后，线程2的操作将不再具备原子性，其他线程又可以竞争锁了。

利器代码

    /**
     * 释放分布式锁
     * @param lockKey   锁
     * @param requestId 请求标识
     * @return 是否释放成功
     */
    public boolean unLock(String lockKey, String requestId) {
        Jedis redis = null;
        try {
            redis = redisManager.redisPool.getResource();
            //lua脚本
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            //执行lua脚本
            Object result = redis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
            if (RELEASE_SUCCESS.equals(result)) {
                return true;
            }
            return false;
        } finally {
            if (null != redis) {
                redis.close();
            }
        }
    }

注意这次我们删除KEY时，不仅传进来了key还有value。这时有童鞋要问了，为什么要这么麻烦？解释一下lua脚本就明白了。
首先获取KEY在redis中的value值，如果结果和传入的value相同，则删除KEY，否则返回0；
这里我们比较了value值来保证删除的是当前线程所拥有的锁，而不是别的线程锁。
注意：我们这里采用的是“脚本形式”，获取与删除在一个redis单线程中同时进行，也是保证操作的原子性,确保删除的一定是value值对应的KEY，而不是别的线程设定的KEY

其他问题：

看到这里我们发现似乎还有一个问题：我们为了防止死锁而设置了KEY的过期时间，可是我们如果不能确保业务代码一定能在KEY过期之前操作完毕，那么就会出现获取锁的线程还没释放锁，其他线程就已经抢到锁，这样就不能保证原子操作，从而不能保证数据的一致性等。这里我们很难想到一个合适的KEY过期时间，而避免这种问题。有没有更好的解决办法呢？答案是肯定的，有。

工具

 /**
     * @Description: 获取锁剩余时间
     * @author: wz
     * @date: 2019/8/2 14:17
     */
    public Long getLockExpire(String key) {
        Jedis redis = null;
        try {
            redis = redisManager.redisPool.getResource();
            return redis.ttl(key);
        } finally {
            if (null != redis) {
                redis.close();
            }
        }
    }

    /**
     * @Description: 设置key 新的过期时间
     * @author: wz
     * @date: 2019/8/2 11:23
     */
    public boolean setLockExpire(String key, int newExpire) {
        Jedis redis = null;
        try {
            redis = redisManager.redisPool.getResource();
            return redis.expire(key, newExpire) > 0;
        } finally {
            if (null != redis) {
                redis.close();
            }
        }
    }

守护线程代码

 //守护线程
    public class DeliveryDaeThread implements Runnable {
        private boolean openDaeThread = true;
        public void open() {
            openDaeThread = true;
        }
        //关闭守护线程
        public void close() {
            openDaeThread = false;
        }
        @Override
        public void run() {
            while (openDaeThread) {
                try {
                    System.out.println("wait...");
                    //在过期时间之前 2s 续命
                    if (getLockExpire() <= 2) {
                        System.out.println("=============>续命");
                        if (openDaeThread) {
                            setLockExpire();
                        }
                    }
                    Thread.sleep(1000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

核心加锁代码

 /**
     * @Description: 对修改redis记录前进行加锁, 如果不能立即获取到锁，则循环等待
     * @author: wz
     * @date: 2019/8/2 14:58
     */
    public DeliveryDaeThread lock(String lockId) throws InterruptedException {
        //尝试获取分布式锁的超时时间 1min
        int timeout = 100;
        boolean locked = this.lockDeliveryRecord(lockId);
        //没取到锁 继续尝试取锁
        while (!locked) {
            Thread.sleep(600);
            Thread thread = Thread.currentThread();
            logger.info(thread.getId() + "--" + thread.getName() + " 尝试获取分布式锁");
            locked = this.lockDeliveryRecord(lockId);
            timeout--;
            //超时
            if (timeout <= 0) {
                break;
            }
        }
        //如果获取到锁，开启守护线程
        if (locked) {
            DeliveryDaeThread daeThread = new DeliveryDaeThread();
            Thread thread = new Thread(daeThread);
            thread.setDaemon(true);
            thread.start();
            return daeThread;
        }
        //没有获取到锁
        return null;
    }

我们可以看到，通过getLockExpire方法可以获取到KEY剩余过期时间,setLockExpire方法可以设置KEY新的过期时间。于是我们可以理下思路：我们可以在线程获取到KEY锁之后开启一个守护线程，这个线程在单位时间就去获取KEY的过期时间，如果即将过期，则给KEY“续命”增加过期时间。注意：这里我们依然享有锁的原子性，其他线程在等待，不会去影响KEY的过期时间。由于获得锁的线程和守护线程在同一个进程，一旦线程出问题停下守护线程也会停下，锁到了超时的时候，不能继续续命，也就自动释放了。
于是我们可以保证在拥有锁的线程执行完业务代码释放锁之前，任何线程不能获取锁。

还有问题？

我们注意到在JDK线程锁里有会有方法设置线程等待超时时间，如果我们要求redis分布式锁的性能就不得不考虑：拥有锁的线程执行完业务代码释放锁之前，任何线程不能获取锁。这个条件造成的阻塞。想像一个极端条件：如果一个线程超久不释放锁那么整个分布式系统等待获取该锁的线程都将等待超久。所以这里我们还需要设置一个等待超时时间，如果线程超时不能获取到锁，难么就会返回null，同时继续执行下面代码。当然超时时间需要根据具体业务代码来设定一个比较中肯合适的值。

结语

终于到结语了啊！笔者此刻的感受就是肩膀超级酸有木有啊！！！从发现问题到解决问题(1d)再到码字写简书(一动不动2h)
遇到问题要充满惊喜，多思考，多动手，最终解决掉，嗯。

将来有打算利用springAOP将redis分布式锁设计成注解模式减少与业务代码的耦合，当然那都是后话了。
如果还有些不明白可以移步到此处:分布式锁
以上。