偷偷摘套内射激情视频,久久精品99国产国产精,中文字幕无线乱码人妻,中文在线中文a,性爽19p

分布式鎖，原來這么簡單！

原創(chuàng) 精選
作者： 蔡柱梁
數(shù)據(jù)庫 Redis
現(xiàn)在的服務往往都是多節(jié)點，在一些特定的場景下容易產(chǎn)生并發(fā)問題，比如扣減庫存，送完即止活動，中臺的批量導入（有唯一校驗要求）等等。這時，我們可以通過分布式鎖解決這些問題。

作者 | 蔡柱梁

審校 | 重樓

目錄

  1. 分布式鎖介紹
  2. 如何實現(xiàn)分布式鎖
  3. 實現(xiàn)分布式鎖

1 分布式鎖介紹

現(xiàn)在的服務往往都是多節(jié)點，在一些特定的場景下容易產(chǎn)生并發(fā)問題，比如扣減庫存，送完即止活動，中臺的批量導入（有唯一校驗要求）等等。這時，我們可以通過分布式鎖解決這些問題。

2 如何實現(xiàn)分布式鎖

實現(xiàn)的方式有很多種，如：

  • 基于 MySQL 等數(shù)據(jù)庫實現(xiàn)
  • 基于 ZooKeeper 實現(xiàn)
  • 基于 Redis 實現(xiàn)不管采用什么技術棧實現(xiàn)，但是邏輯流程都是大體不差的。下面是筆者自己在工作中基于Redis 實踐過的流程圖：

3 實現(xiàn)分布式鎖

其實可以不用自己手寫，現(xiàn)在有一個中間件Redisson 相當好用，十分推薦。這里的實現(xiàn)更多是用于學習。

3.1 Redis 是單節(jié)點的情況下實現(xiàn)的分布式鎖

需要使用分布式鎖的業(yè)務代碼如下：

package com.example.demo.test.utils;

import com.example.demo.utils.RedisLockUtil;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

/**
 * @author CaiZhuliang
 * @date 2023/8/31
 */
@Slf4j
@SpringBootTest
public class RedisLockUtilTest {
 @Autowired
 private RedisLockUtil redisLockUtil;

 @Test
 public void simpleLockTest() {
 String key = "redis:lock:" + System.currentTimeMillis();
 boolean result = redisLockUtil.lock(key, 8_000L);
 if (result) {
 try {
 // do something
 } catch (Exception e) {
 log.error("simpleLockTest - 系統(tǒng)異常!", e);
 } finally {
 boolean unlock = redisLockUtil.unlock(key);
 if (!unlock) {
 log.error("simpleLockTest - 釋放鎖失敗，key : {}", key);
 }
 }
 }
 }
}

分布式鎖工具類代碼如下：

package com.example.demo.utils;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author CaiZhuliang
 * @date 2023/8/31
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisLockUtil {
 private static final ScheduledExecutorService EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(50,
 new BasicThreadFactory.Builder()
 .namingPattern("redisLockUtil-schedule-pool-%d")
 .daemon(true)
 .build());

 private static final ThreadLocal<String> THREAD_LOCAL = new ThreadLocal<>();

 private final RedisTemplate<String, String> redisTemplate;

 /**
 * 釋放鎖
 * <p>必須和RedisLockUtil#simpleLock是同一個線程</p>
 * @param key 需要釋放鎖的key
 * @return true-成功 false-失敗
 */
 public boolean releaseSimpleLock(String key) {
 String token = THREAD_LOCAL.get();
 try {
 String remoteToken = redisTemplate.opsForValue().get(key);
 if (!token.equals(remoteToken)) {
 // 當前線程不再持有鎖
 return false;
 }
 // 是自己持有鎖才能釋放
 return Boolean.TRUE.equals(redisTemplate.delete(key));
 } catch (Exception e) {
 log.error("非cluster模式簡單分布式鎖 - 釋放鎖發(fā)生異常，key : {}", key, e);
 return false;
 } finally {
 THREAD_LOCAL.remove();
 }
 }

 /**
 * 這個方法不考慮Redis的集群架構，不考慮腦裂問題，當只有一個Redis來考慮。
 * @param key 需要上鎖的key
 * @param expireTime 過期時間，單位：毫秒
 * @return true-成功 false-失敗
 */
 public boolean simpleLock(String key, Long expireTime) {
 if (StringUtils.isBlank(key)) {
 log.warn("非cluster模式簡單分布式鎖 - key is blank");
 return false;
 }
 if (null == expireTime || expireTime <= 0) {
 expireTime = 0L;
 }
 String token = UUID.randomUUID().toString();
 // 續(xù)約周期，單位納秒
 long renewPeriod = expireTime / 2 * 1000_000;
 try {
 // 設置鎖
 Boolean result = redisTemplate.opsForValue().setIfAbsent(key, token, expireTime, TimeUnit.MILLISECONDS);
 if (Boolean.FALSE.equals(result)) {
 return false;
 }
 // 上鎖成功后將令牌綁定當前線程
 THREAD_LOCAL.set(token);
 if (renewPeriod > 0) {
 // 續(xù)約任務
 renewTask(key, token, expireTime, renewPeriod);
 }
 return true;
 } catch (Exception e) {
 log.error("非cluster模式簡單分布式鎖 - 上鎖失敗。", e);
 THREAD_LOCAL.remove();
 return false;
 }
 }

 /**
 * 鎖續(xù)約任務
 * @param key 需要續(xù)命的key
 * @param token 成功獲鎖的線程持有的令牌
 * @param expireTime 過期時間，單位：毫秒
 * @param renewPeriod 續(xù)約周期，單位：納秒
 */
 private void renewTask(String key, String token, long expireTime, long renewPeriod) {
 EXECUTOR_SERVICE.schedule(() -> {
 ValueOperations<String, String> valueOperator = redisTemplate.opsForValue();
 String val = valueOperator.get(key);
 if (token.equals(val)) {
 // 是自己持有鎖才能續(xù)約
 try {
 Boolean result = valueOperator.setIfPresent(key, token, expireTime, TimeUnit.MILLISECONDS);
 if (Boolean.TRUE.equals(result)) {
 // 續(xù)約成功
 log.debug("非cluster模式簡單分布式鎖 - 鎖續(xù)約成功，key : {}", key);
 // 開啟下一次續(xù)約任務
 renewTask(key, token, expireTime, renewPeriod);
 } else {
 log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約失敗，key : {}", key);
 }
 } catch (Exception e) {
 // 這里異常是拋不出去的，所以需要 catch 打印
 log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約發(fā)生異常，key : {}", key, e);
 }
 } else {
 log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約失敗，不再持有token，key : {}", key);
 }
 }, renewPeriod, TimeUnit.NANOSECONDS);
 }
}

這就是一個最簡單的實現(xiàn)方式。不過這里存在著許多問題：

  • 續(xù)約任務

這里判斷是否持有令牌和續(xù)約這兩個動作不在同一個事務里，可能發(fā)生覆蓋現(xiàn)象。假設A線程判斷自己持有令牌，但是一直沒有請求 Redis 導致鎖過期。B線程成功獲鎖，這時A線程往下執(zhí)行 Redis 請求，結果A線程搶了B線程的鎖。

  • 釋放鎖

這里判斷是否持有令牌和刪除key這兩個動作不在同一個事務里，可能出現(xiàn)誤刪現(xiàn)象。假設A線程現(xiàn)在要釋放鎖，通過了令牌判斷，準備刪除 key 但是還沒執(zhí)行。這時 key 過期了，B線程成功獲鎖。接著A線程執(zhí)行刪除 key 導致了 B 線程的鎖被刪除。

因此，判斷持有令牌與續(xù)約/刪除key這兩個動作是需要原子性的，我們可以通過 lua 來實現(xiàn)。

擴展，了解管道與 lua 的區(qū)別

  • pipeline（多用于命令簡單高效，無關聯(lián)的場景）

優(yōu)點：使用簡單，有效減少網(wǎng)絡IO

缺點：本質還是發(fā)送命令請求Redis 服務，如果效率過低，就會阻塞 Redis，導致 Redis 無法處理其他請求

  • lua（多用于命令復雜，命令間有關聯(lián)的場景）

優(yōu)點：

  1. Redis 支持 lua 腳本，Redis 服務執(zhí)行 lua 的同時是可以處理別的請求的，不會產(chǎn)生阻塞
  2. 命令都在腳本中，有效減少網(wǎng)絡IO
  3. 具有原子性

缺點：

有一定的學習成本

3.1.1 使用 lua 進行優(yōu)化

RedisLockUtil 代碼如下:

package com.example.demo.utils;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author CaiZhuliang
 * @date 2023/8/31
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisLockUtil {
 private static final ScheduledExecutorService EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(50,
 new BasicThreadFactory.Builder()
 .namingPattern("redisLockUtil-schedule-pool-%d")
 .daemon(true)
 .build());

 private static final ThreadLocal<String> THREAD_LOCAL = new ThreadLocal<>();
 private static final String SUCCESS = "1";
 /**
 * 允許當前token續(xù)約
 */
 private static final Integer CAN_RENEW = 0;
 /**
 * 記錄token的狀態(tài)，0-可以續(xù)約，其他情況均不能續(xù)約
 */
 private static final Map<String, Integer> TOKEN_STATUS = Maps.newConcurrentMap();

 private final RedisTemplate<String, String> redisTemplate;

 /**
 * 釋放鎖，這個方法與 com.example.demo.utils.RedisLockUtil#simpleLock(java.lang.String, java.lang.Long) 配對。
 * <p>必須和RedisLockUtil#simpleLock是同一個線程</p>
 * @param key 需要釋放鎖的key
 * @return true-成功 false-失敗
 */
 public boolean releaseSimpleLock(String key) {
 String token = THREAD_LOCAL.get();
 if (null != token) {
 TOKEN_STATUS.put(token, 1);
 }
 try {
 String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
 "then redis.call('expire', KEYS[1], 0) return '1' end " +
 "return '0'";
 DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
 String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token);
 log.info("非cluster模式簡單分布式鎖 - 釋放key: {}, result : {}, token : {}", key, result, token);
 return SUCCESS.equals(result);
 } catch (Exception e) {
 log.error("非cluster模式簡單分布式鎖 - 釋放鎖發(fā)生異常，key : {}", key, e);
 return false;
 } finally {
 THREAD_LOCAL.remove();
 if (null != token) {
 TOKEN_STATUS.remove(token);
 }
 }
 }

 /**
 * 簡單分布式鎖實現(xiàn)，續(xù)約周期是 expireTime 的一半。舉個例子， expireTime = 8000，那么鎖續(xù)約將會是每 4000 毫秒續(xù)約一次
 * <p>這個方法不考慮Redis的集群架構，不考慮腦裂問題，當只有一個 Redis來考慮。</p>
 * <p>這個方法使用 com.example.demo.utils.RedisLockUtil#releaseSimpleLock(java.lang.String) 來釋放鎖</p>
 * @param key 需要上鎖的key
 * @param expireTime 過期時間，單位：毫秒
 * @return true-成功 false-失敗
 */
 public boolean simpleLock(String key, Long expireTime) {
 if (StringUtils.isBlank(key)) {
 log.warn("非cluster模式簡單分布式鎖 - key is blank");
 return false;
 }
 if (null == expireTime || expireTime <= 0) {
 expireTime = 0L;
 }
 // 續(xù)約周期，單位納秒
 long renewPeriod = expireTime / 2 * 1000_000;
 try {
 String token = System.currentTimeMillis() + ":" + UUID.randomUUID();
 // 設置鎖
 Boolean result = redisTemplate.opsForValue().setIfAbsent(key, token, expireTime, TimeUnit.MILLISECONDS);
 if (Boolean.FALSE.equals(result)) {
 return false;
 }
 log.info("非cluster模式簡單分布式鎖 - 上鎖成功，key : {}, token : {}", key, token);
 // 上鎖成功后將令牌綁定當前線程
 THREAD_LOCAL.set(token);
 TOKEN_STATUS.put(token, 0);
 if (renewPeriod > 0) {
 // 續(xù)約任務
 renewTask(key, token, expireTime, renewPeriod);
 }
 return true;
 } catch (Exception e) {
 log.error("非cluster模式簡單分布式鎖 - 上鎖發(fā)生異常，key : {}", key, e);
 String token = THREAD_LOCAL.get();
 if (StringUtils.isNotBlank(token)) {
 if (!releaseSimpleLock(key)) {
 log.warn("非cluster模式簡單分布式鎖 - 釋放鎖發(fā)生失敗，key : {}, token : {}", key, token);
 }
 }
 return false;
 }
 }

 /**
 * 鎖續(xù)約任務
 * @param key 需要續(xù)命的key
 * @param token 成功獲鎖的線程持有的令牌
 * @param expireTime 過期時間，單位：毫秒
 * @param renewPeriod 續(xù)約周期，單位：納秒
 */
 private void renewTask(String key, String token, long expireTime, long renewPeriod) {
 if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
 EXECUTOR_SERVICE.schedule(() -> {
 if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
 try {
 String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
 "then " +
 " if (redis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2])) " +
 " then return '1' else return redis.call('get', KEYS[1]) end " +
 "end " +
 "return redis.call('get', KEYS[1])";
 DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
 String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token, String.valueOf(expireTime));
 if (SUCCESS.equals(result)) {
 // 續(xù)約成功
 log.debug("非cluster模式簡單分布式鎖 - 鎖續(xù)約成功，key : {}", key);
 // 開啟下一次續(xù)約任務
 renewTask(key, token, expireTime, renewPeriod);
 } else {
 // 打印下 result，看下是否因為不再持有令牌導致的續(xù)約失敗
 log.warn("非cluster模式簡單分布式鎖 - 鎖續(xù)約失敗，key : {}, token : {}, result : {}", key, token, result);
 }
 } catch (Exception e) {
 // 這里異常是拋不出去的，所以需要 catch 打印
 log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約發(fā)生異常，key : {}", key, e);
 }
 }
 }, renewPeriod, TimeUnit.NANOSECONDS);
 }
 }
}

這里還有一個問題：如果redis.call('get', KEYS[1]) == ARGV[1] 成立，但是執(zhí)行redis.call('expire', KEYS[1], 0) 失敗，怎么辦？我這里已經(jīng)執(zhí)行了THREAD_LOCAL.remove()，想重復釋放是不可能的了，但是我這里不能不 remove 或者僅當 Redis 釋放鎖成功才 remove，這樣存在內存泄漏的風險。要怎么處理呢？

這是優(yōu)化后的代碼：

package com.example.demo.utils;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author CaiZhuliang
 * @date 2023/8/31
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisSimpleLockUtil {
 private static final ScheduledExecutorService EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(50,
 new BasicThreadFactory.Builder()
 .namingPattern("redisSimpleLockUtil-schedule-pool-%d")
 .daemon(true)
 .build());

 private static final ThreadLocal<String> THREAD_LOCAL_TOKEN = new ThreadLocal<>();
 private static final String SUCCESS = "1";
 /**
 * 允許當前token續(xù)約
 */
 private static final Integer CAN_RENEW = 0;
 /**
 * 記錄token的狀態(tài)，0-可以續(xù)約，其他情況均不能續(xù)約
 */
 private static final Map<String, Integer> TOKEN_STATUS = Maps.newConcurrentMap();

 private final RedisTemplate<String, String> redisTemplate;

 /**
 * 釋放鎖
 * <p>必須和 RedisSimpleLockUtil#lock 是同一個線程</p>
 * @param key key 需要釋放鎖的key
 * @param token 持有的令牌
 * @return true-成功 false-失敗
 */
 public boolean releaseLock(String key, String token) {
 if (StringUtils.isBlank(token)) {
 return false;
 }
 TOKEN_STATUS.put(token, 1);
 try {
 String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
 "then redis.call('expire', KEYS[1], 0) return '1' end " +
 "return '0'";
 DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
 String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token);
 log.info("非cluster模式簡單分布式鎖 - 釋放key: {}, result : {}, token : {}", key, result, token);
 if (SUCCESS.equals(result)) {
 return true;
 }
 String remoteToken = redisTemplate.opsForValue().get(key);
 if (token.equals(remoteToken)) {
 log.warn("非cluster模式簡單分布式鎖 - 釋放鎖失敗，key : {}, token : {}", key, token);
 return false;
 }
 return true;
 } catch (Exception e) {
 log.error("非cluster模式簡單分布式鎖 - 釋放鎖發(fā)生異常，key : {}, token : {}", key, token, e);
 return false;
 } finally {
 THREAD_LOCAL_TOKEN.remove();
 TOKEN_STATUS.remove(token);
 }
 }

 /**
 * 簡單分布式鎖實現(xiàn)，續(xù)約周期是 expireTime 的一半。舉個例子， expireTime = 8000，那么鎖續(xù)約將會是每 4000 毫秒續(xù)約一次
 * <p>這個方法不考慮Redis的集群架構，不考慮腦裂問題，當只有一個Redis來考慮。</p>
 * @param key 需要上鎖的key
 * @param expireTime 過期時間，單位：毫秒
 * @return 上鎖成功返回令牌，失敗則返回空串
 */
 public String lock(String key, Long expireTime) {
 if (StringUtils.isBlank(key)) {
 log.warn("非cluster模式簡單分布式鎖 - key is blank");
 return StringUtils.EMPTY;
 }
 if (null == expireTime || expireTime <= 0) {
 expireTime = 0L;
 }
 // 續(xù)約周期，單位納秒
 long renewPeriod = expireTime * 500_000;
 try {
 String token = System.currentTimeMillis() + ":" + UUID.randomUUID();
 // 設置鎖
 Boolean result = redisTemplate.opsForValue().setIfAbsent(key, token, expireTime, TimeUnit.MILLISECONDS);
 if (Boolean.FALSE.equals(result)) {
 return StringUtils.EMPTY;
 }
 log.info("非cluster模式簡單分布式鎖 - 上鎖成功，key : {}, token : {}", key, token);
 // 上鎖成功后將令牌綁定當前線程
 THREAD_LOCAL_TOKEN.set(token);
 TOKEN_STATUS.put(token, 0);
 if (renewPeriod > 0) {
 // 續(xù)約任務
 log.info("非cluster模式簡單分布式鎖 - 添加續(xù)約任務，key : {}, token : {}, renewPeriod : {}納秒", key, token, renewPeriod);
 renewTask(key, token, expireTime, renewPeriod);
 }
 return token;
 } catch (Exception e) {
 String token = THREAD_LOCAL_TOKEN.get();
 log.error("非cluster模式簡單分布式鎖 - 上鎖發(fā)生異常，key : {}, token : {}", key, token, e);
 return StringUtils.isBlank(token) ? StringUtils.EMPTY : token;
 }
 }

 /**
 * 鎖續(xù)約任務
 * @param key 需要續(xù)命的key
 * @param token 成功獲鎖的線程持有的令牌
 * @param expireTime 過期時間，單位：毫秒
 * @param renewPeriod 續(xù)約周期，單位：納秒
 */
 private void renewTask(String key, String token, long expireTime, long renewPeriod) {
 try {
 EXECUTOR_SERVICE.schedule(() -> {
 try {
 String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
 "then " +
 " if (redis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2])) " +
 " then return '1' else return redis.call('get', KEYS[1]) end " +
 "end " +
 "return redis.call('get', KEYS[1])";
 DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
 String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token, String.valueOf(expireTime));
 if (SUCCESS.equals(result)) {
 // 續(xù)約成功
 log.debug("非cluster模式簡單分布式鎖 - 鎖續(xù)約成功，key : {}, token : {}", key, token);
 // 這里加判斷是為了減少定時任務
 if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
 // 開啟下一次續(xù)約任務
 renewTask(key, token, expireTime, renewPeriod);
 }
 } else {
 // 這里加判斷是為了防止誤打印warn日志
 if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
 log.warn("非cluster模式簡單分布式鎖 - 鎖續(xù)約失敗，key : {}, token : {}, result : {}", key, token, result);
 }
 }
 } catch (Exception e) {
 // 這里異常是拋不出去的，所以需要 catch 打印
 log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約發(fā)生異常，key : {}, token : {}", key, token, e);
 }
 }, renewPeriod, TimeUnit.NANOSECONDS);
 } catch (Exception e) {
 log.error("非cluster模式簡單分布式鎖 - 添加鎖續(xù)約任務發(fā)生異常，key : {}, token : {}", key, token, e);
 }
 }
}
package com.example.demo.utils;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author CaiZhuliang
 * @date 2023/8/31
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisSimpleLockUtil {
 private static final ScheduledExecutorService EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(50,
 new BasicThreadFactory.Builder()
 .namingPattern("redisSimpleLockUtil-schedule-pool-%d")
 .daemon(true)
 .build());

 private static final ThreadLocal<String> THREAD_LOCAL_TOKEN = new ThreadLocal<>();
 private static final String SUCCESS = "1";
 /**
 * 允許當前token續(xù)約
 */
 private static final Integer CAN_RENEW = 0;
 /**
 * 記錄token的狀態(tài)，0-可以續(xù)約，其他情況均不能續(xù)約
 */
 private static final Map<String, Integer> TOKEN_STATUS = Maps.newConcurrentMap();

 private final RedisTemplate<String, String> redisTemplate;

 /**
 * 釋放鎖
 * <p>必須和 RedisSimpleLockUtil#lock 是同一個線程</p>
 * @param key key 需要釋放鎖的key
 * @param token 持有的令牌
 * @return true-成功 false-失敗
 */
 public boolean releaseLock(String key, String token) {
 if (StringUtils.isBlank(token)) {
 return false;
 }
 TOKEN_STATUS.put(token, 1);
 try {
 String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
 "then redis.call('expire', KEYS[1], 0) return '1' end " +
 "return '0'";
 DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
 String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token);
 log.info("非cluster模式簡單分布式鎖 - 釋放key: {}, result : {}, token : {}", key, result, token);
 if (SUCCESS.equals(result)) {
 return true;
 }
 String remoteToken = redisTemplate.opsForValue().get(key);
 if (token.equals(remoteToken)) {
 log.warn("非cluster模式簡單分布式鎖 - 釋放鎖失敗，key : {}, token : {}", key, token);
 return false;
 }
 return true;
 } catch (Exception e) {
 log.error("非cluster模式簡單分布式鎖 - 釋放鎖發(fā)生異常，key : {}, token : {}", key, token, e);
 return false;
 } finally {
 THREAD_LOCAL_TOKEN.remove();
 TOKEN_STATUS.remove(token);
 }
 }

 /**
 * 簡單分布式鎖實現(xiàn)，續(xù)約周期是 expireTime 的一半。舉個例子， expireTime = 8000，那么鎖續(xù)約將會是每 4000 毫秒續(xù)約一次
 * <p>這個方法不考慮Redis的集群架構，不考慮腦裂問題，當只有一個Redis來考慮。</p>
 * @param key 需要上鎖的key
 * @param expireTime 過期時間，單位：毫秒
 * @return 上鎖成功返回令牌，失敗則返回空串
 */
 public String lock(String key, Long expireTime) {
 if (StringUtils.isBlank(key)) {
 log.warn("非cluster模式簡單分布式鎖 - key is blank");
 return StringUtils.EMPTY;
 }
 if (null == expireTime || expireTime <= 0) {
 expireTime = 0L;
 }
 // 續(xù)約周期，單位納秒
 long renewPeriod = expireTime * 500_000;
 try {
 String token = System.currentTimeMillis() + ":" + UUID.randomUUID();
 // 設置鎖
 Boolean result = redisTemplate.opsForValue().setIfAbsent(key, token, expireTime, TimeUnit.MILLISECONDS);
 if (Boolean.FALSE.equals(result)) {
 return StringUtils.EMPTY;
 }
 log.info("非cluster模式簡單分布式鎖 - 上鎖成功，key : {}, token : {}", key, token);
 // 上鎖成功后將令牌綁定當前線程
 THREAD_LOCAL_TOKEN.set(token);
 TOKEN_STATUS.put(token, 0);
 if (renewPeriod > 0) {
 // 續(xù)約任務
 log.info("非cluster模式簡單分布式鎖 - 添加續(xù)約任務，key : {}, token : {}, renewPeriod : {}納秒", key, token, renewPeriod);
 renewTask(key, token, expireTime, renewPeriod);
 }
 return token;
 } catch (Exception e) {
 String token = THREAD_LOCAL_TOKEN.get();
 log.error("非cluster模式簡單分布式鎖 - 上鎖發(fā)生異常，key : {}, token : {}", key, token, e);
 return StringUtils.isBlank(token) ? StringUtils.EMPTY : token;
 }
 }

 /**
 * 鎖續(xù)約任務
 * @param key 需要續(xù)命的key
 * @param token 成功獲鎖的線程持有的令牌
 * @param expireTime 過期時間，單位：毫秒
 * @param renewPeriod 續(xù)約周期，單位：納秒
 */
 private void renewTask(String key, String token, long expireTime, long renewPeriod) {
 try {
 EXECUTOR_SERVICE.schedule(() -> {
 try {
 String lua = "if (redis.call('get', KEYS[1]) == ARGV[1]) " +
 "then " +
 " if (redis.call('set', KEYS[1], ARGV[1], 'PX', ARGV[2])) " +
 " then return '1' else return redis.call('get', KEYS[1]) end " +
 "end " +
 "return redis.call('get', KEYS[1])";
 DefaultRedisScript<String> luaScript = new DefaultRedisScript<>(lua, String.class);
 String result = redisTemplate.execute(luaScript, Lists.newArrayList(key), token, String.valueOf(expireTime));
 if (SUCCESS.equals(result)) {
 // 續(xù)約成功
 log.debug("非cluster模式簡單分布式鎖 - 鎖續(xù)約成功，key : {}, token : {}", key, token);
 // 這里加判斷是為了減少定時任務
 if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
 // 開啟下一次續(xù)約任務
 renewTask(key, token, expireTime, renewPeriod);
 }
 } else {
 // 這里加判斷是為了防止誤打印warn日志
 if (CAN_RENEW.equals(TOKEN_STATUS.get(token))) {
 log.warn("非cluster模式簡單分布式鎖 - 鎖續(xù)約失敗，key : {}, token : {}, result : {}", key, token, result);
 }
 }
 } catch (Exception e) {
 // 這里異常是拋不出去的，所以需要 catch 打印
 log.error("非cluster模式簡單分布式鎖 - 鎖續(xù)約發(fā)生異常，key : {}, token : {}", key, token, e);
 }
 }, renewPeriod, TimeUnit.NANOSECONDS);
 } catch (Exception e) {
 log.error("非cluster模式簡單分布式鎖 - 添加鎖續(xù)約任務發(fā)生異常，key : {}, token : {}", key, token, e);
 }
 }
}

下面是并發(fā)單元測試代碼：

@Test
 public void concurrencyTest() {
 String[] nums = {"1", "2", "3", "4", "5"};
 List<CompletableFuture<Void>> list = Lists.newArrayListWithExpectedSize(100);
 for (int i = 0; i < 50; i++) {
 CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
 for (int count = 0; count < 10; count++) {
 int random = new Random().nextInt(100) % 5;
 String key = "test_" + nums[random];
 while (true) {
 String token = redisSimpleLockUtil.lock(key, 3_000L);
 if (StringUtils.isNotBlank(token)) {
 log.info("concurrencyTest - key : {}", key);
 try {
 Thread.sleep(new Random().nextInt(1500));
 } catch (Exception e) {
 log.error("concurrencyTest - 發(fā)生異常, key : {}", key, e);
 } finally {
 boolean unlock = redisSimpleLockUtil.releaseLock(key, token);
 if (!unlock) {
 log.error("concurrencyTest - 釋放鎖失敗，key : {}", key);
 }
 }
 break;
 }
 }
 }
 });
 list.add(future);
 }
 CompletableFuture<?>[] futures = new CompletableFuture[list.size()];
 list.toArray(futures);
 CompletableFuture.allOf(futures).join();
 }

3.2 紅鎖

一般公司使用Redis 時都不可能是單節(jié)點的，要么主從+哨兵架構，要么就是 cluster 架構。面對集群，我們不得不思考如何應對腦裂這個問題。而 Redlock 是Redis官方網(wǎng)站給出的解決方案。

下面看下針對這兩種集群架構的處理方式：

  1. 主從+哨兵

通過訪問哨兵獲取當前 master 節(jié)點，統(tǒng)計票數(shù)，超過半數(shù)的 master 節(jié)點就是真的 master。我們可以對比我們成功上鎖的節(jié)點是否是真的 master node，從而避免腦裂問題。

  1. cluster
  2. 上鎖需要在集群中半數(shù)以上的 master 操作成功了才算成功。

3.2.1 紅鎖的問題

紅鎖通過過半原則來規(guī)避腦裂，但是這就讓我們不得不考慮訪問節(jié)點的等待超時時間應該要多長。而且，也會降低Redis 分布式鎖的吞吐量。如果有半數(shù)節(jié)點不可用，那么分布式鎖也將變得不可用。因此，實際使用中，我們還要結合自己實際的業(yè)務場景來權衡要不要用紅鎖或者修改實現(xiàn)方案。

作者介紹

蔡柱梁，51CTO社區(qū)編輯，從事Java后端開發(fā)8年，做過傳統(tǒng)項目廣電BOSS系統(tǒng)，后投身互聯(lián)網(wǎng)電商，負責過訂單，TMS，中間件等。


責任編輯：華軒 來源： 51CTO
分布式鎖Redis
相關推薦

2021-06-10 06:57:39

Redis存儲數(shù)據(jù)庫

2021-02-02 16:37:25

Redis分布式鎖

2018-10-28 17:54:00

分布式事務數(shù)據(jù)

2023-10-10 18:26:58

分布式鎖緩存

2021-11-11 07:47:03

Redis分布式鎖

2019-06-19 15:40:06

分布式鎖RedisJava

2021-04-19 05:42:51

Mmap文件系統(tǒng)

2022-03-08 07:22:48

Redis腳本分布式鎖

2018-07-17 08:14:22

分布式分布式鎖方位

2022-08-04 08:45:50

Redisson分布式鎖工具

2019-02-26 09:51:52

分布式鎖RedisZookeeper

2021-07-16 07:57:34

ZooKeeperCurator源碼

2018-11-27 16:17:13

分布式鎖Tomcat

2021-11-26 06:43:19

Java分布式鎖

2023-11-01 14:49:07

2021-07-06 08:37:29

Redisson分布式鎖

2021-10-25 10:21:59

ZK分布式鎖ZooKeeper

2022-01-06 10:58:07

Redis數(shù)據(jù)分布式鎖

2023-08-21 19:10:34

Redis分布式鎖

2017-10-24 11:28:23

Zookeeper分布式鎖架構
點贊
收藏

51CTO技術棧公眾號