redis实现分布式锁的两种方式

　　使用数据库写锁、synchronized、ReentrantLock等都可以实现对于数据的线程安全控制。但这些都属于排它锁（或者你也可以认为是悲观锁）范畴，会造成一定的阻塞，无法满足快速响应的要求。

---

基于【高并发抢购防止超卖】的案例。　　　　　　　　　　　　　　　　　　　　

我们使用redis的两种不同方式，实现分布式锁。　　　　　　　　　　　　　　

【阅读前提：您对redis中的watch、事务、setnx有一定的了解】

---

一、基于watch机制

　　这种相当于是乐观锁的实现方式，乐观的以为没人和我抢。乐观锁适用于“读多写少”的场景。此处仅作为练习使用。方式二才是通常用法。

package qianggou;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import comm.Value;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Transaction;

/**
 * 测试抢购案例 
 *
 */
public class RedisTest {

    public static void main(String[] args) {
        final String watchkeys = "watchkeys";
        ExecutorService excutor = Executors.newFixedThreadPool(20);//开启最多20个线程的线程池，相当于真实场景中的限流
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        JedisPool jedisPool = new JedisPool(jedisPoolConfig, "aliyun", 6379, 5000, Value.PASSWORD);//Value.PASSWORD是你的redis设置密码，这里我使用接口常量封装了
        
        final Jedis jedis = jedisPool.getResource();
        jedis.set(watchkeys, "0");
        jedis.del("setsucc","setfail");
        jedis.close();
        
        
        for(int i=0;i<1000;i++) {
            excutor.execute(new MyRunnable(jedisPool));
        }
        excutor.shutdown();
    }
}
class MyRunnable implements Runnable{

    String watchkeys = "watchkeys";
    JedisPool jedisPool = null;
    
    public MyRunnable(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
    }
    @Override
    public void run() {
        Jedis jedis = jedisPool.getResource();
        jedis.watch(watchkeys);//监听watchkeys
        String val = jedis.get(watchkeys);
        jedis.set(watchkeys, "1");
        int valint = Integer.valueOf(val);
        String userinfo = UUID.randomUUID().toString();
        if(valint < 10) {
            
            Transaction tx = jedis.multi();
            tx.incr(watchkeys);//更改watchkeys的值。
            List<Object> exec = tx.exec();//如果watchkeys被其他线程修改了，则抢购失败
            if(exec !=null && exec.size()>0) {
                System.out.println("用户【"+userinfo+"】抢购成功，当前抢购成功人数为："+(valint+1));
                jedis.sadd("setsucc", userinfo);
                jedis.close();
                return;
            }
        }
        jedis.sadd("setfail", userinfo);
        jedis.close();
    }
}

二、基于setnx

　　这种相当于是悲观锁的实现方式，没有获取到锁则抢购失败。（其实真实的悲观锁是会进行等待阻塞的）

package qianggou;

import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import comm.Value;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class RedisTest02 {

    
    public static void main(String[] args) {
        final String SAIL_KEY = "sailkey";
        ExecutorService excutor = Executors.newFixedThreadPool(20);
        
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        JedisPool jedisPool = new JedisPool(jedisPoolConfig, "aliyun", 6379, 5000, Value.PASSWORD);
        
        final Jedis jedis = jedisPool.getResource();
        jedis.del(SAIL_KEY);
        jedis.set(SAIL_KEY, "0");
        jedis.del("setsucc","setfail","LOCK");
        jedis.close();
        
        for(int i=0;i<100;i++) {
            excutor.execute(new MyRunnable2(jedisPool));
        }
        excutor.shutdown();
    }

}
class MyRunnable2 implements Runnable{

    String LOCK = "LOCK";
    String SAIL_KEY = "sailkey";
    JedisPool jedisPool  = null;
    
    public MyRunnable2(JedisPool jedisPool ) {
        this.jedisPool = jedisPool;
    }
    @Override
    public void run() {
        
        String userinfo = UUID.randomUUID().toString();
        
        Jedis jedis = jedisPool.getResource();

        Long lock = jedis.setnx(LOCK, userinfo);
        if(lock>0 ? false : true) {//这一步在spring中封装为RedisTemplate了
            return;
        }    
        
        ScheduledExecutorService schedul = Executors.newScheduledThreadPool(1);
        try {
            jedis.expire(LOCK, 2);//初始化锁定时间
            
            //异常点一：设置锁的初始锁定时间，如果2秒钟之内方法未执行完，则通过以下定时器为锁续命
            schedul.schedule(new Runnable() {
                @Override
                public void run() {
                    jedis.expire(LOCK, 2); 
                }
            }, 1, TimeUnit.SECONDS);//定时每1秒为锁续期
            
            int num = jedis.incr(SAIL_KEY).intValue();
            
            if(num<=10) {
                System.out.println("用户【"+userinfo+"】抢购成功，当前抢购成功人数为："+num);
            }
        }finally {
            schedul.shutdownNow();//关闭所有定时子进程
            //异常点二：如果执行到这一步，突然卡住了Thread.sleep()，LOCK时间也到期了。
            //别的线程就可以重新生成LOCK这把锁，防止以下代码删除了别的线程的LOCK
            if(userinfo.equals(jedis.get(LOCK))) {//防止误删
                jedis.del(LOCK); //方式一
            }
            jedis.close();//先删除再关闭
        }
    }
    
}

三、使用redisson进行简化

　　方案二中的加锁看起来过于繁琐了，接下来使用redisson对其加锁续期过程进行简化。我们使用一下spring中的RedisTemplate（你也可以不使用）。

　　pom中需要引入引用：

<dependency>
        <groupId>org.springframework.data</groupId>
        <artifactId>spring-data-redis</artifactId>
        <version>1.7.7.RELEASE</version>
 </dependency>
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.11.3</version>
</dependency>

　　spring配置文件：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans.xsd
     http://www.springframework.org/schema/context
     http://www.springframework.org/schema/context/spring-context.xsd">

    <bean id="redisPoolConfig"
        class="redis.clients.jedis.JedisPoolConfig"></bean>

    <bean id="JedisConnectionFactory"
        class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
        <property name="hostName" value="127.0.0.1"></property>
        <property name="port" value="6379"></property>
        <property name="password" value="123456"></property>
        <property name="poolConfig" ref="redisPoolConfig"></property>
    </bean>
    
    <bean id="stringRedisSerializer"
        class="org.springframework.data.redis.serializer.StringRedisSerializer" />
    <bean id="redisTemplate"
        class="org.springframework.data.redis.core.RedisTemplate">
        <property name="connectionFactory" ref="JedisConnectionFactory" />
        <property name="keySerializer" ref="stringRedisSerializer" />
        <property name="valueSerializer" ref="stringRedisSerializer" />
    </bean>

</beans>

　　测试代码：

package test;

import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;

public class Miaosha2 {
    
    public static void main(String[] args) {
        
        ApplicationContext ac = new ClassPathXmlApplicationContext("applicationContext-redis.xml");
        RedisTemplate<String, String> redisTemplate = ac.getBean(RedisTemplate.class);
        //数据初始化
        redisTemplate.opsForValue().set("sails", "0");//初始化库存已售数量
        redisTemplate.delete("LOCK");
        
        System.out.println(redisTemplate);
        
        ExecutorService excutor = Executors.newFixedThreadPool(20);
        for(int i=0;i<100;i++) {    
            excutor.execute(new MyRunnable2(redisTemplate));
        }
        excutor.shutdown();
    
    }

}
class MyRunnable2 implements Runnable{
    RedisTemplate<String, String> redisTemplate;
    public MyRunnable2(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
    @Override
    public void run() {
        String userInfo = UUID.randomUUID().toString();
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        config.useSingleServer().setPassword("123456");

        RedissonClient redisson = Redisson.create(config);
        RLock lock = redisson.getLock("LOCK");
        
        try {
            lock.lock();//获取锁，获取不到则结束

        redisTemplate.opsForValue().set("sails", String.valueOf(num));
            Long num = redisTemplate.opsForValue().increment("sails", 1);
            if(num <= 10) {
                System.out.println("用户【" + userInfo + "】抢购成功，当前抢购成功人数为：" + num);
            }
        } finally {
            lock.unlock();//释放锁
            redisson.shutdown();//需要关闭redisson连接，否则会占用redis连接资源
        }
        
    }
    
}

 四、原子性优化

　　以上方式[包含redisson方式]外存在一个很严重的额问题是，设置锁和设置超时时间的代码是非原子性操作。

　　我们想象一个场景如果设置完锁之后系统异常宕机，但是没有设置超时时间，这导致的后果就是：锁永远无法释放了。

　　为了保证这两步的原子性操作===>，

　　在redis2.6.12版本之前，可以使用LUA脚本方式：　

public class Lua {
    
    //加锁脚本(索引从1开始)
    private static final String SCRIPT_TRYLOCK = "if redis.call('setnx',KEYS[1],ARGV[1]) ==1 "
            + " then redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end";
    
    /**
     * 使用Lua脚本，尝试获取分布式锁
     */
    public static boolean tryLockLua(Jedis jedis, String lockkey, String lockValue, int expireTime) {
        
        int result = (int) jedis.eval(SCRIPT_TRYLOCK, 1, lockkey,String.valueOf(expireTime));//设置锁
        if(result == 1) {
            return true ;
        }
        return false;
    }
}

　　在redis2.6.12版本及其之后，对set命令进行了增强，同时作者也不建议使用setnx命令了，这个命令也没存在的必要了，后续版本可能会将其删除：

/**
     * 命令：SET key value NX PX miliseconds
     *    如：SET key value NX PX 1000
     */
    public static boolean tryLock(Jedis jedis, String lockkey, String lockValue, int expireTime) {
        
        String result = jedis.set(lockkey, lockValue,"NX","PX",expireTime);
        if("OK".equals(result)) {
            return true ;
        }
        return false;
    }