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;
    }