如何利用Redis分布式锁实现控制并发操作
时间：2021-05-20 09:28:59 | 栏目：Redis | 点击：次
redis命令解释
说道Redis的分布式锁都是通过setNx命令结合getset来实现的，在讲之前我们先了解下setNx和getset的意思，在redis官网是这样解释的
注：redis的命令都是原子操作
SETNX key value
将 key 的值设为 value ，当且仅当 key 不存在。
若给定的 key 已经存在，则 SETNX 不做任何动作。
SETNX 是『SET if Not eXists』(如果不存在，则 SET)的简写。
可用版本：
1.0.0+
时间复杂度：
O(1)
返回值：
设置成功，返回 1 。
设置失败，返回 0 。
redis> EXISTS job # job 不存在 (integer) 0 redis> SETNX job "programmer" # job 设置成功 (integer) 1 redis> SETNX job "code-farmer" # 尝试覆盖 job ，失败 (integer) 0 redis> GET job # 没有被覆盖 "programmer"
GETSET key value
将给定 key 的值设为 value ，并返回 key 的旧值(old value)。
当 key 存在但不是字符串类型时，返回一个错误。
可用版本：
1.0.0+
时间复杂度：
O(1)
返回值：
返回给定 key 的旧值。
当 key 没有旧值时，也即是， key 不存在时，返回 nil 。
redis> GETSET db mongodb # 没有旧值，返回 nil (nil) redis> GET db "mongodb" redis> GETSET db redis # 返回旧值 mongodb "mongodb" redis> GET db "redis"
代码示例
注意：为了让分布式锁的算法更稳键些，持有锁的客户端在解锁之前应该再检查一次自己的锁是否已经超时，再去做DEL操作，因为可能客户端因为某个耗时的操作而挂起，操作完的时候锁因为超时已经被别人获得，这时就不必解锁了。
我们看下代码涉及以下几个类，这里有关业务逻辑相关的只定义了方法没有具体实现，关键是学习思路
OrderBiz.java
/**
* 使用redis锁来控制并发抢单
* @author fuyuwei
*/
public class OrderBiz {
public int createOrder(){
// 下单之前的参数、合法性校验这里就不在演示
OrderLock<Boolean> orderLock = new RedisOrderLock<Boolean>("pro-12345678901");
boolean isSyn = orderLock.isSyn(new OrderLockBiz<Boolean>(){
@Override
public Boolean createOrder() {
// 省去创建订单逻辑
return null;
}
});
if(!isSyn){
BizLogger.info("创建订单失败");
}
return 0;
}
}
OrderLock.java
public interface OrderLock<T> {
public boolean isSyn(OrderLockBiz<T> orderBiz);
}
OrderLockBiz.java
public interface OrderLockBiz<T> {
public T createOrder();
}
RedisOrderLock.java
public class RedisOrderLock<T> implements OrderLock<T> {
// 锁等待超时，防止线程饥饿，永远没有入锁执行代码的机会
public static final long timeout = 10000;//ms
// 锁持有超时，防止线程在入锁以后，无限的执行下去，让锁无法释放
public static final long expireMsecs = 10000;// ms
public String lockKey = "orderLockKey";
public Jedis jedis;
private static volatile JedisPool jedisPool;
public RedisOrderLock(String lockKey) {
this.lockKey = lockKey;
}
/**
* 初始化redis
* @return
*/
public Jedis getInstance() {
if(jedisPool == null) {
synchronized(RedisOrderLock.class) {
if(jedisPool == null) {
JedisPoolConfig config = new JedisPoolConfig();
config.setMaxIdle(100);
jedisPool = new JedisPool(config,"localhost",6379, 3000,"test");
}
}
}
return jedisPool.getResource();
}
/**
* 线程安全的业务逻辑处理
*/
@Override
public boolean isSyn(OrderLockBiz<T> orderBiz) {
jedis = this.getInstance();
try {
// 获取到锁
if(acquire(jedis)){
// 执行创建订单逻辑
orderBiz.createOrder();
}else{
BizLogger.info("waiting other thread creating");
}
} catch (Exception e) {
BizLogger.error(e,"acquire lock failre");
}finally{
// 解锁
this.releaseLock(jedis);
}
return false;
}
/**
* accqure lock
* @param jedis
* @return
* @throws InterruptedException
*/
public synchronized boolean acquire(Jedis jedis){
boolean locked = false;
while(timeout > 0){
long expires = System.currentTimeMillis() + expireMsecs + 1;
// 10秒之后锁到期
String expiresStr = String.valueOf(expires);
// 获取到锁
if(jedis.setnx(lockKey, expiresStr) == 1){
locked = true;
return locked;
}
// 没有获取到锁
String oldValue = jedis.get(lockKey);
// expireMsecs（10秒）锁的有效期内无法进入if判断，如果锁超时了
if(oldValue != null
&& Long.parseLong(oldValue) < System.currentTimeMillis()){
// 如果锁超时重新设置
String oldValue_ = jedis.getSet(lockKey, expiresStr);
// 值相同说明是同一个线程的操作，获取锁成功
if(Long.valueOf(oldValue_) == Long.valueOf(oldValue)){
locked = true;
}else{
// 被其他线程抢先获取锁
locked = false;
}
}
// 锁没有超时，继续等待
return false;
}
}
/**
* 释放锁
* @param jedis
*/
public synchronized void releaseLock(Jedis jedis){
try {
long current = System.currentTimeMillis();
// 避免删除非自己获取得到的锁
if (current < Long.valueOf(jedis.get(lockKey)))
jedis.del(lockKey);
} catch (Exception e) {
e.printStackTrace();
}finally{
// 把用完的连接放到连接池汇中供其他线程调用
jedisPool.returnResource(jedis);
}
}
}