使用数据库写锁、synchronized、ReentrantLock等都可以实现对于数据的线程安全控制。但这些都属于排它锁(或者你也可以认为是悲观锁)范畴,会造成一定的阻塞,无法满足快速响应的要求。
基于【高并发抢购防止超卖】的案例。
我们使用redis的两种不同方式,实现分布式锁。
【阅读前提:您对redis中的watch、事务、setnx有一定的了解】
一、基于watch机制
这种相当于是乐观锁的实现方式,乐观的以为没人和我抢。乐观锁适用于“读多写少”的场景。此处仅作为练习使用。方式二才是通常用法。
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1 package qianggou; 2 3 import java.util.List; 4 import java.util.UUID; 5 import java.util.concurrent.ExecutorService; 6 import java.util.concurrent.Executors; 7 8 import comm.Value; 9 import redis.clients.jedis.Jedis; 10 import redis.clients.jedis.JedisPool; 11 import redis.clients.jedis.JedisPoolConfig; 12 import redis.clients.jedis.Transaction; 13 14 /** 15 * 测试抢购案例 16 * 17 */ 18 public class RedisTest { 19 20 public static void main(String[] args) { 21 final String watchkeys = "watchkeys"; 22 ExecutorService excutor = Executors.newFixedThreadPool(20);//开启最多20个线程的线程池,相当于真实场景中的限流 23 JedisPoolConfig jedisPoolConfig = new JedisPoolConfig(); 24 JedisPool jedisPool = new JedisPool(jedisPoolConfig, "aliyun", 6379, 5000, Value.PASSWORD);//Value.PASSWORD是你的redis设置密码,这里我使用接口常量封装了 25 26 final Jedis jedis = jedisPool.getResource(); 27 jedis.set(watchkeys, "0"); 28 jedis.del("setsucc","setfail"); 29 jedis.close(); 30 31 32 for(int i=0;i<1000;i++) { 33 excutor.execute(new MyRunnable(jedisPool)); 34 } 35 excutor.shutdown(); 36 } 37 } 38 class MyRunnable implements Runnable{ 39 40 String watchkeys = "watchkeys"; 41 JedisPool jedisPool = null; 42 43 public MyRunnable(JedisPool jedisPool) { 44 this.jedisPool = jedisPool; 45 } 46 @Override 47 public void run() { 48 Jedis jedis = jedisPool.getResource(); 49 jedis.watch(watchkeys);//监听watchkeys 50 String val = jedis.get(watchkeys); 51 jedis.set(watchkeys, "1"); 52 int valint = Integer.valueOf(val); 53 String userinfo = UUID.randomUUID().toString(); 54 if(valint < 10) { 55 56 Transaction tx = jedis.multi(); 57 tx.incr(watchkeys);//更改watchkeys的值。 58 List<Object> exec = tx.exec();//如果watchkeys被其他线程修改了,则抢购失败 59 if(exec !=null && exec.size()>0) { 60 System.out.println("用户【"+userinfo+"】抢购成功,当前抢购成功人数为:"+(valint+1)); 61 jedis.sadd("setsucc", userinfo); 62 jedis.close(); 63 return; 64 } 65 } 66 jedis.sadd("setfail", userinfo); 67 jedis.close(); 68 } 69 }
二、基于setnx
这种相当于是悲观锁的实现方式,没有获取到锁则抢购失败。(其实真实的悲观锁是会进行等待阻塞的)
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1 package qianggou; 2 3 import java.util.UUID; 4 import java.util.concurrent.ExecutorService; 5 import java.util.concurrent.Executors; 6 import java.util.concurrent.ScheduledExecutorService; 7 import java.util.concurrent.TimeUnit; 8 9 import comm.Value; 10 import redis.clients.jedis.Jedis; 11 import redis.clients.jedis.JedisPool; 12 import redis.clients.jedis.JedisPoolConfig; 13 14 public class RedisTest02 { 15 16 17 public static void main(String[] args) { 18 final String SAIL_KEY = "sailkey"; 19 ExecutorService excutor = Executors.newFixedThreadPool(20); 20 21 JedisPoolConfig jedisPoolConfig = new JedisPoolConfig(); 22 JedisPool jedisPool = new JedisPool(jedisPoolConfig, "aliyun", 6379, 5000, Value.PASSWORD); 23 24 final Jedis jedis = jedisPool.getResource(); 25 jedis.del(SAIL_KEY); 26 jedis.set(SAIL_KEY, "0"); 27 jedis.del("setsucc","setfail","LOCK"); 28 jedis.close(); 29 30 for(int i=0;i<100;i++) { 31 excutor.execute(new MyRunnable2(jedisPool)); 32 } 33 excutor.shutdown(); 34 } 35 36 } 37 class MyRunnable2 implements Runnable{ 38 39 String LOCK = "LOCK"; 40 String SAIL_KEY = "sailkey"; 41 JedisPool jedisPool = null; 42 43 public MyRunnable2(JedisPool jedisPool ) { 44 this.jedisPool = jedisPool; 45 } 46 @Override 47 public void run() { 48 49 String userinfo = UUID.randomUUID().toString(); 50 51 Jedis jedis = jedisPool.getResource(); 52 53 Long lock = jedis.setnx(LOCK, userinfo); 54 if(lock>0 ? false : true) {//这一步在spring中封装为RedisTemplate了 55 return; 56 } 57 58 ScheduledExecutorService schedul = Executors.newScheduledThreadPool(1); 59 try { 60 jedis.expire(LOCK, 2);//初始化锁定时间 61 62 //异常点一:设置锁的初始锁定时间,如果2秒钟之内方法未执行完,则通过以下定时器为锁续命 63 schedul.schedule(new Runnable() { 64 @Override 65 public void run() { 66 jedis.expire(LOCK, 2); 67 } 68 }, 1, TimeUnit.SECONDS);//定时每1秒为锁续期 69 70 int num = jedis.incr(SAIL_KEY).intValue(); 71 72 if(num<=10) { 73 System.out.println("用户【"+userinfo+"】抢购成功,当前抢购成功人数为:"+num); 74 } 75 }finally { 76 schedul.shutdownNow();//关闭所有定时子进程 77 //异常点二:如果执行到这一步,突然卡住了Thread.sleep(),LOCK时间也到期了。 78 //别的线程就可以重新生成LOCK这把锁,防止以下代码删除了别的线程的LOCK 79 if(userinfo.equals(jedis.get(LOCK))) {//防止误删 80 jedis.del(LOCK); //方式一 81 } 82 jedis.close();//先删除再关闭 83 } 84 } 85 86 }
三、使用redisson进行简化
方案二中的加锁看起来过于繁琐了,接下来使用redisson对其加锁续期过程进行简化。我们使用一下spring中的RedisTemplate(你也可以不使用)。
pom中需要引入引用:
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1 <dependency> 2 <groupId>org.springframework.data</groupId> 3 <artifactId>spring-data-redis</artifactId> 4 <version>1.7.7.RELEASE</version> 5 </dependency> 6 <dependency> 7 <groupId>org.redisson</groupId> 8 <artifactId>redisson</artifactId> 9 <version>3.11.3</version> 10 </dependency>
spring配置文件:
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1 <?xml version="1.0" encoding="UTF-8"?> 2 <beans xmlns="http://www.springframework.org/schema/beans" 3 xmlns:context="http://www.springframework.org/schema/context" 4 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 5 xsi:schemaLocation="http://www.springframework.org/schema/beans 6 http://www.springframework.org/schema/beans/spring-beans.xsd 7 http://www.springframework.org/schema/context 8 http://www.springframework.org/schema/context/spring-context.xsd"> 9 10 <bean id="redisPoolConfig" 11 class="redis.clients.jedis.JedisPoolConfig"></bean> 12 13 <bean id="JedisConnectionFactory" 14 class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory"> 15 <property name="hostName" value="127.0.0.1"></property> 16 <property name="port" value="6379"></property> 17 <property name="password" value="123456"></property> 18 <property name="poolConfig" ref="redisPoolConfig"></property> 19 </bean> 20 21 <bean id="stringRedisSerializer" 22 class="org.springframework.data.redis.serializer.StringRedisSerializer" /> 23 <bean id="redisTemplate" 24 class="org.springframework.data.redis.core.RedisTemplate"> 25 <property name="connectionFactory" ref="JedisConnectionFactory" /> 26 <property name="keySerializer" ref="stringRedisSerializer" /> 27 <property name="valueSerializer" ref="stringRedisSerializer" /> 28 </bean> 29 30 </beans>
测试代码:
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1 package test; 2 3 import java.util.UUID; 4 import java.util.concurrent.ExecutorService; 5 import java.util.concurrent.Executors; 6 7 import org.redisson.Redisson; 8 import org.redisson.api.RLock; 9 import org.redisson.api.RedissonClient; 10 import org.redisson.config.Config; 11 import org.springframework.context.ApplicationContext; 12 import org.springframework.context.support.ClassPathXmlApplicationContext; 13 import org.springframework.data.redis.core.RedisTemplate; 14 15 public class Miaosha2 { 16 17 public static void main(String[] args) { 18 19 ApplicationContext ac = new ClassPathXmlApplicationContext("applicationContext-redis.xml"); 20 RedisTemplate<String, String> redisTemplate = ac.getBean(RedisTemplate.class); 21 //数据初始化 22 redisTemplate.opsForValue().set("sails", "0");//初始化库存已售数量 23 redisTemplate.delete("LOCK"); 24 25 System.out.println(redisTemplate); 26 27 ExecutorService excutor = Executors.newFixedThreadPool(20); 28 for(int i=0;i<100;i++) { 29 excutor.execute(new MyRunnable2(redisTemplate)); 30 } 31 excutor.shutdown(); 32 33 } 34 35 } 36 class MyRunnable2 implements Runnable{ 37 RedisTemplate<String, String> redisTemplate; 38 public MyRunnable2(RedisTemplate<String, String> redisTemplate) { 39 this.redisTemplate = redisTemplate; 40 } 41 @Override 42 public void run() { 43 String userInfo = UUID.randomUUID().toString(); 44 Config config = new Config(); 45 config.useSingleServer().setAddress("redis://127.0.0.1:6379"); 46 config.useSingleServer().setPassword("123456"); 47 48 RedissonClient redisson = Redisson.create(config); 49 RLock lock = redisson.getLock("LOCK"); 50 51 try { 52 lock.lock();//获取锁,获取不到则结束 53 54 redisTemplate.opsForValue().set("sails", String.valueOf(num)); 55 Long num = redisTemplate.opsForValue().increment("sails", 1); 56 if(num <= 10) { 57 System.out.println("用户【" + userInfo + "】抢购成功,当前抢购成功人数为:" + num); 58 } 59 } finally { 60 lock.unlock();//释放锁 61 redisson.shutdown();//需要关闭redisson连接,否则会占用redis连接资源 62 } 63 64 } 65 66 }
四、原子性优化
以上方式[包含redisson方式]外存在一个很严重的额问题是,设置锁和设置超时时间的代码是非原子性操作。
我们想象一个场景如果设置完锁之后系统异常宕机,但是没有设置超时时间,这导致的后果就是:锁永远无法释放了。
为了保证这两步的原子性操作===>,
在redis2.6.12版本之前,可以使用LUA脚本方式:
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1 public class Lua { 2 3 //加锁脚本(索引从1开始) 4 private static final String SCRIPT_TRYLOCK = "if redis.call('setnx',KEYS[1],ARGV[1]) ==1 " 5 + " then redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end"; 6 7 /** 8 * 使用Lua脚本,尝试获取分布式锁 9 */ 10 public static boolean tryLockLua(Jedis jedis, String lockkey, String lockValue, int expireTime) { 11 12 int result = (int) jedis.eval(SCRIPT_TRYLOCK, 1, lockkey,String.valueOf(expireTime));//设置锁 13 if(result == 1) { 14 return true ; 15 } 16 return false; 17 } 18 }
在redis2.6.12版本及其之后,对set命令进行了增强,同时作者也不建议使用setnx命令了,这个命令也没存在的必要了,后续版本可能会将其删除:
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
1 /** 2 * 命令:SET key value NX PX miliseconds 3 * 如:SET key value NX PX 1000 4 */ 5 public static boolean tryLock(Jedis jedis, String lockkey, String lockValue, int expireTime) { 6 7 String result = jedis.set(lockkey, lockValue,"NX","PX",expireTime); 8 if("OK".equals(result)) { 9 return true ; 10 } 11 return false; 12 }