从原理到实践分析 Redis 分布式锁的多种实现方案

SETNX key value

void setnxCommand(client *c) {
    robj *o;
    int nx = c->argc == 3; /* 如果参数个数为 3，说明设置 NX（key 不存在才设置） */
    long long expire = 0; /* 默认不设置过期时间 */
    int retval;
    if (nx) {
        /* NX 模式下检查 key 是否已经存在 */
        if (lookupKeyWrite(c->db,c->argv[1]) != NULL) {
            addReply(c,shared.czero);
            return;
        }
    } else {
        /* XX 模式下检查 key 是否不存在 */
        if (lookupKeyWrite(c->db,c->argv[1]) == NULL) {
            addReply(c,shared.czero);
            return;
        }
    }
    /* 尝试将字符串型或整型数字转换为 long long 型数字 */
    if (getTimeoutFromObjectOrReply(c,c->argv[3],&expire,UNIT_SECONDS)
        != C_OK) return;
    /* 值为空则返回错误 */
    if (checkStringLength(c,c->argv[2]->ptr,sdslen(c->argv[2]->ptr)) != C_OK)
        return;
    /* 尝试将键值对插入到数据库中 */
    o = createStringObject(c->argv[2]->ptr,sdslen(c->argv[2]->ptr));
    retval = dictAdd(c->db->dict,c->argv[1],o);
    if (retval == DICT_OK) {
        incrRefCount(o);
        /* 设置过期时间 */
        if (expire) setExpire(c->db,c->argv[1],mstime()+expire);
        server.dirty++;
        addReply(c, shared.cone);
    } else {
        decrRefCount(o);
        addReply(c, shared.czero);
    }
}

@Service
public class ProductService {
    private final RedisTemplate<String, String> redisTemplate;
    @Autowired
    public ProductService(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
    /**
     * 扣减库存
     *
     * @param productId 商品ID
     * @param quantity  数量
     * @return true 扣减成功，false 扣减失败
     */
    public boolean decreaseStock(String productId, int quantity) {
        String lockKey = "stock_" + productId;
        while (true) {
            Boolean lockResult = redisTemplate.opsForValue().setIfAbsent(lockKey, "", 10, TimeUnit.SECONDS);
            if (lockResult != null && lockResult) {
                try {
                    String stockKey = "product_" + productId;
                    String stockStr = redisTemplate.opsForValue().get(stockKey);
                    if (StringUtils.isEmpty(stockStr)) {
                        // 库存不存在或已过期
                        return false;
                    }
                    int stock = Integer.parseInt(stockStr);
                    if (stock < quantity) {
                        // 库存不足
                        return false;
                    }
                    int newStock = stock - quantity;
                    redisTemplate.opsForValue().set(stockKey, String.valueOf(newStock));
                    return true;
                } finally {
                    redisTemplate.delete(lockKey);
                }
            }
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

SETNXEX key seconds value

void setnxexCommand(client *c) {
    robj *key = c->argv[1], *val = c->argv[3];
    long long expire = strtoll(c->argv[2]->ptr,NULL,10);
    expire *= 1000;
    if (getExpire(c,key) != -1) {
        addReply(c, shared.czero);
        return;
    }
    setKey(c,c->db,key,val,LOOKUP_NOTOUCH|LOOKUP_EX|LOOKUP_NX,0,0,NULL);
    if (c->flags & CLIENT_MULTI) {
        addReply(c, shared.cone);
        return;
    }
    server.dirty++;
    if (expire) setExpire(c,c->db,key,mstime()+expire);
    addReply(c, shared.cone);
    notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
    notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);
}

@Component
public class StockService {
    private final Logger logger = LoggerFactory.getLogger(StockService.class);
    private final String LOCK_KEY_PREFIX = "stock:lock:";
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 扣减库存
     * @param productId 商品ID
     * @param num 扣减数量
     */
    public boolean reduceStock(Long productId, int num) {
        // 构造锁的key
        String lockKey = LOCK_KEY_PREFIX + productId;
        // 构造锁的value，这里使用当前线程的ID
        String lockValue = String.valueOf(Thread.currentThread().getId());
        try {
            // 尝试获取锁，设置过期时间为10秒
            Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 10L, TimeUnit.SECONDS);
            if (!locked) {
                // 获取锁失败，等待10秒后重新尝试获取锁
                Thread.sleep(10000);
                return reduceStock(productId, num);
            }
            // 获取锁成功，执行扣减库存代码
            // TODO ... 扣减库存代码
            return true;
        } catch (InterruptedException e) {
            logger.error("Failed to acquire stock lock", e);
            return false;
        } finally {
            // 释放锁
            if (lockValue.equals(redisTemplate.opsForValue().get(lockKey))) {
                redisTemplate.delete(lockKey);
            }
        }
    }
}

@Component
public class StockService {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    /**
     * 扣减库存
     *
     * @param stockId 库存 ID
     * @param num 扣减数量
     * @return 是否扣减成功
     */
    public boolean reduceStock(String stockId, int num) throws InterruptedException {
        // 构造锁的名称
        String lockKey = "stock_lock_" + stockId;
        // 获取当前线程 ID
        String threadId = String.valueOf(Thread.currentThread().getId());
        try {
            // 使用 SETNXEX 命令申请锁
            Boolean lockResult = redisTemplate.opsForValue().setIfAbsent(lockKey, threadId, 30, TimeUnit.SECONDS);
            if (!lockResult) {
                // 如果获取锁失败，则等待一段时间后重试
                Thread.sleep(10000);
                return reduceStock(stockId, num);
            }
            // 设置锁的过期时间为 30 秒，并启动一个定时任务续租锁
            ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
            scheduledExecutorService.scheduleAtFixedRate(() -> {
                Long expireResult = redisTemplate.getExpire(lockKey);
                if (expireResult < 10) {
                    redisTemplate.expire(lockKey, expireResult + 10, TimeUnit.SECONDS);
                }
            }, 10, 10, TimeUnit.SECONDS);
            // TODO：执行业务逻辑，例如扣减库存
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 释放锁
            if (threadId.equals(redisTemplate.opsForValue().get(lockKey))) {
                redisTemplate.delete(lockKey);
            }
        }
        return false;
    }
}