导言

---

后端程序员多多少少都与各种各样的“锁”打过交道，如： 电商网站在下单时对某个商品减库存，必须加锁防止多个客户同时下单导致库存量错误； 抢分享红包时必须对红包加锁防止用户抢到的金额超过红包上限等。

对于单体或部署在单台服务器上的应用而言，对某种资源加锁相对简单，使用语言本身的并发控制机制或是文件锁即可实现。然而为了提高响应速度以及防止单点，当前的应用常常会在多台服务器上部署多个实例，这时就需要进行分布式的并发控制了。

一种实现分布式锁的思路利用了redis的setnx命令，除了这种思路以外，Redis在2.6推出了脚本功能，允许开发者使用Lua语言编写脚本传到Redis中执行。由于redis是单线程的，单次执行的lua脚本可以认为是原子的，redis的java客户端redisson就利用了这一特性实现了分布式锁。

redisson的加锁原理

---

redisson加锁的逻辑在这里，如果释放锁的时间不是-1，则直接试着加锁，否则认为该锁的超时时间为无穷大，在成功获得锁之后每隔一段时间刷新锁的过期时间。

   private RFuture<Boolean> tryAcquireOnceAsync(long leaseTime, 
    TimeUnit unit,
    final long threadId) {
        if (leaseTime != -1) {
            return tryLockInnerAsync(leaseTime,
              unit,
              threadId, 
              RedisCommands.EVAL_NULL_BOOLEAN);
        }
        RFuture<Boolean> ttlRemainingFuture 
                  = tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS,
                             TimeUnit.SECONDS, 
                             threadId, 
                             RedisCommands.EVAL_NULL_BOOLEAN);

        ttlRemainingFuture.addListener(new FutureListener<Boolean>() {
            @Override
            public void operationComplete(Future<Boolean> future) throws Exception {
                if (!future.isSuccess()) {
                    return;
                }

                Boolean ttlRemaining = future.getNow();
                // lock acquired
                if (ttlRemaining) {
                    scheduleExpirationRenewal(threadId);
                }
            }
        });
        return ttlRemainingFuture;
    }

加锁的过程主要是靠下面这段lua脚本：

    <T> RFuture<T> tryLockInnerAsync(long leaseTime, 
           TimeUnit unit,
           long threadId, 
           RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        return commandExecutor.evalWriteAsync(getName(),
                  LongCodec.INSTANCE, 
                  command,
                  "if (redis.call('exists', KEYS[1]) == 0) then " +
                      "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                      "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "return redis.call('pttl', KEYS[1]);",
                  Collections.<Object>singletonList(getName()), 
                  internalLockLeaseTime,
                  getLockName(threadId));
    }

上面的KEYS[1]为加锁时指定的锁的名称，argv[1]是internalLockLeaseTime，argv[2]是thread_id。这段逻辑为：如果锁不存在，则hset 锁名称 线程id 1，直接返回。 如果hexists 锁名称 线程id返回1，则说明该客户端已经加过锁了，则将自己线程id对应的值增加1，并重新设置该key的超时时间，实现可重入。否则说明该锁已经被其他客户端获取，返回锁还有多久过期。如果加锁失败，客户端可以选择利用redis的pub/sub机制监听一个channel，在监听到锁被释放后尝试重新加锁，或者返会失败让客户端决定下一步操作。

解锁时KEY[1]为锁名称，KEY[2]为关注的队列名称，ARGV[1]为解锁消息，ARGV[2]为超时时间，ARGV[3]为线程:

   protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return commandExecutor.evalWriteAsync(getName(),
                  LongCodec.INSTANCE,
                  RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; " +
                "end;" +
                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                    "return nil;" +
                "end; " +
                "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                "if (counter > 0) then " +
                    "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                    "return 0; " +
                "else " +
                    "redis.call('del', KEYS[1]); " +
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; "+
                "end; " +
                "return nil;",
                Arrays.<Object>asList(getName(), 
                           getChannelName()),
                           LockPubSub.unlockMessage, 
                           internalLockLeaseTime, getLockName(threadId));
    }

解锁的时候如果锁已经超时，则通知其他等待获取该锁的客户端并返回1。如果该锁没有被当前客户端获得，则什么都不做直接返回null。如果当期客户端拥有锁，则为了可重入，先减少一次当前线程获取锁的次数，如果unlock次数小于lock次数，则重新设置锁的超时时间，返回0，如果锁已经被彻底释放，则删除掉锁，并通知其他等待获取锁的客户端，并返回1表示解锁成功。

总体而言，redisson的锁防止了由于客户端死掉而导致的锁永久被占用，还支持可重入锁，功能上还是比较完备的。

封装redisson

---

redisson的分布式锁从功能上说已经够用了，为了方便使用以及与spring集成，我们需要再将其做一次封装，由于redis有集群、单机、sentinel几种模式，我们将加锁功能抽象起来，提供统一的接口。话不多说，我们之间上代码：

/** 
  * Created by millions on 2017/1/6.
 * 分布式锁 
 */
public interface DistributedLock {    
/**    
  * 使用分布式锁，获取锁后执行callback指定的业务逻辑，使用锁默认超时时间。     
  * 
  * @param callback 完成回调     
  * @return  
  */ 
   <T> T lock(DistributedLockCallback<T> callback);  

  /**
    * 使用分布式锁，获取锁后执行callback指定的业务逻辑，使用锁默认超时时间。     
    *
    * @param callback  
    * @param leaseTime 锁超时时间。超时后自动释放锁。   
    * @param timeUnit 超时时间单位    
    * @return
    */
    <T> T lock(DistributedLockCallback<T> callback, long leaseTime, TimeUnit timeUnit);
}

我们将加锁后需要的逻辑封装到callback中：

/** * Created by millions on 2017/1/6. * 分布式锁回调接口 */
public interface DistributedLockCallback<T> { 
   T process();   
   String getLockName();
}

实现单机redis的分布式锁:

public class SingleDistributedLock implements DistributedLock {
    private static final long DEFAULT_TIMEOUT = 5;
    private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.SECONDS;
    private RedissonClient redissonClient;

    public SingleDistributedLock(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    public SingleDistributedLock() {
    }

    public <T> T lock(DistributedLockCallback<T> callback) {
        return lock(callback, DEFAULT_TIMEOUT, DEFAULT_TIME_UNIT);
    }

    public <T> T lock(DistributedLockCallback<T> callback, long leaseTime,
        TimeUnit timeUnit) {
        RLock lock = null;

        try {
            lock = redissonClient.getLock(callback.getLockName());
            lock.lock(leaseTime, timeUnit);

            return callback.process();
        } finally {
            if (null != lock) {
                lock.unlock();
            }
        }
    }

    public void setRedissonClient(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }
}

为了方便的和spring整合，我们实现spring的FactoryBean接口

public class DistributedLockFactoryBean implements FactoryBean<DistributedLock> {
    private static final Logger logger 
     = LoggerFactory.getLogger(DistributedLockFactoryBean.class);

    private RedissonClient client;

    private LockModel lockModel;

    private DistributedLock lock;

    public DistributedLockFactoryBean(String modelName) {
        lockModel = LockModel.fromString(modelName);
    }

    @PostConstruct
    public void init() {
        logger.debug("DistributedLockFactoryBean is initializing!");
        //从classpath下读取RedissionClient配置
        InputStream inputStream = null;
        Config config = null;
        try {
            inputStream = DistributedLockFactoryBean
            .class
            .getClassLoader()
            .getResourceAsStream("redis-config.json");
            config = Config.fromJSON(inputStream);
        } catch (IOException e) {
            logger.error("读取redission配置失败");
        } finally {
            if (null != inputStream) {
                try {
                    inputStream.close();
                } catch (IOException e) {
                    logger.info("", e);
                }
            }
        }
        Assert.notNull(config);
        //初始化client
        client = Redisson.create(config);
    }

    @PreDestroy
    public void destroy() {
        logger.debug("DistributedLockFactoryBean will be destroyed!");
        client.shutdown();
    }

    public DistributedLock getObject() throws Exception {
        Class<? extends DistributedLock> clz = lockModel.lockClass();
        lock = clz.newInstance();
        Method setRedissonClient
            = clz.getMethod("setRedissonClient", RedissonClient.class);
        if (null != setRedissonClient) {
            setRedissonClient.invoke(lock, client);
            return lock;
        }
        throw new InvalidObjectException(clz.getSimpleName()+"不支持的RedissonClient");
    }

    public Class<?> getObjectType() {
        return DistributedLock.class;
    }

    public boolean isSingleton() {
        return true;
    }

    private enum LockModel {
        SINGLE {
            Class<? extends DistributedLock> lockClass() {
                return SingleDistributedLock.class;
            }
        };

        abstract Class<? extends DistributedLock> lockClass();

        public static LockModel fromString(String modelName) {
            LockModel[] values = LockModel.values();
            for (LockModel model : values) {
                if (model.name().equalsIgnoreCase(modelName)) {
                    return model;
                }
            }
            throw new InvalidParameterException("找不到合适的LockModel");
        }
    }
}

使用时引入jar包，在resource目录下放入redis配置文件，配置好factoryBean 就可以愉快的使用啦~

{
  "singleServerConfig": {
    "idleConnectionTimeout": 10000,
    "pingTimeout": 1000,
    "connectTimeout": 10000,
    "timeout": 3000,
    "retryAttempts": 3,
    "retryInterval": 1500,
    "reconnectionTimeout": 3000,
    "failedAttempts": 3,
    "password": null,
    "subscriptionsPerConnection": 5,
    "clientName": null,
    "address": "redis://192.168.0.133:6379",
    "subscriptionConnectionMinimumIdleSize": 1,
    "subscriptionConnectionPoolSize": 50,
    "connectionMinimumIdleSize": 10,
    "connectionPoolSize": 64,
    "database": 0,
    "dnsMonitoring": false,
    "dnsMonitoringInterval": 5000
  },
  "threads": 0,
  "nettyThreads": 0,
  "codec": null,
  "useLinuxNativeEpoll": false //redisson使用netty 需要指定是否使用epoll模式
}

总结

---

总结一下，redisson利用redis工作线程为单线程，lua脚本能够原子执行的这一特性，利用lua脚本和redis的pub/sub特性实现了分布式、可重入的锁。在redisson的基础上进行了封装，并添加了与spring的整合。这个分布式锁没用什么高大上的原理，代码也写的很屌丝，但是足以支撑一般规模的业务。从这里我想到，很多时候业务上提的一些需求，我们其实都应该多思考一下，将其抽象，整合成一般的工具库，加快自己以及同事的开发效率。从另外一个角度来说，需求方可能也要给码农们稍微充裕一点的时间，多想，多思考，这样才能更好的推进公司的发展。