Redis并发计数工具
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import java.util.Optional;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Redis并发计数工具
*
* @author caochong
* @version 1.0
*/
public class RedisCountLock {
private final AtomicInteger reentryCountAtomic = new AtomicInteger(0);
private final RedisTemplate<String, Integer> redisTemplate;
private final String NUM_KEY;
private int maxWaitNum = 20;
private long retryWaitTime = 100L;
public RedisCountLock(RedisTemplate<String, Integer> redisTemplate, String NUM_KEY) {
this.redisTemplate = redisTemplate;
this.NUM_KEY = NUM_KEY;
}
public RedisCountLock(RedisTemplate<String, Integer> redisTemplate, String NUM_KEY,
int maxWaitNum, long retryWaitTime) {
this.redisTemplate = redisTemplate;
this.NUM_KEY = NUM_KEY;
this.maxWaitNum = maxWaitNum;
this.retryWaitTime = retryWaitTime;
}
/**
* 增加计数
*/
public void incr() throws InterruptedException {
ValueOperations<String, Integer> ops = redisTemplate.opsForValue();
do {
Integer currentVal = ops.get(NUM_KEY);
if (currentVal != null) {
if (currentVal < maxWaitNum) {
long updateVal = Optional.ofNullable(ops.increment(NUM_KEY, 1)).orElse(currentVal + 1L);
if (updateVal - 1 <= currentVal) {
break;
} else {
// System.out.println("补偿：" + currentVal + " - " + updateVal + " - " + reentryCountAtomic.get());
// 补偿redis计数
ops.increment(NUM_KEY, -1);
}
}
Thread.sleep(retryWaitTime + ThreadLocalRandom.current().nextInt(-50, 50));
} else {
ops.setIfAbsent(NUM_KEY, 0);
}
} while (true);
reentryCountAtomic.incrementAndGet(); // 重入次数
}
/**
* 减计数
*/
public void decr() {
int reentryCount = reentryCountAtomic.getAndDecrement(); // 扣减重入次数
if (reentryCount <= 0) {
// 重入补偿
System.out.println("扣减补偿：" + reentryCount + " - " + reentryCountAtomic.get());
reentryCountAtomic.incrementAndGet();
} else {
// 扣成功减计数
ValueOperations<String, Integer> ops = redisTemplate.opsForValue();
ops.increment(NUM_KEY, -1);
}
}
public int reentryCount() {
return reentryCountAtomic.get();
}
}
通过标识分组的分布式重入互斥锁（缺少续期，可自行追加）
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import java.util.concurrent.TimeUnit;
/**
* Redis锁
*
* @author caoch
*/
public class RedisLock {
private static final Logger logger = LoggerFactory.getLogger(RedisLock.class);
private final RedisTemplate<String, Object> redisTemplate;
private final int lockType;
private final String key;
private volatile boolean isOpen = false;
// 互斥锁特有：时间版本
private volatile String lockTypeVersion = "";
private volatile long lockTimeVersion = 0;
private volatile long lockExTime = 0;
public RedisLock(RedisTemplate<String, Object> redisTemplate, String key) {
this.redisTemplate = redisTemplate;
this.key = key;
this.lockType = 0;
}
public RedisLock(RedisTemplate<String, Object> redisTemplate, String key, int lockType) {
this.redisTemplate = redisTemplate;
this.key = key;
this.lockType = lockType;
}
/**
* redis等待锁
*
* @param lockExTime 锁时长，单位：秒
* @param sleepTime 重试间隔，单位：毫秒
* @param maxWaitTime 最大等待时长，单位：毫秒
* @return 是否获取到锁
* @throws InterruptedException 中断异常
*/
public boolean tryLock(long lockExTime, long sleepTime,
long maxWaitTime) throws InterruptedException {
if (this.lockType != 0) {
throw new RuntimeException("Lock method not supported: tryLock");
}
long keyTime = System.currentTimeMillis();
ValueOperations<String, Object> ops = redisTemplate.opsForValue();
do {
Object obj = ops.get(key);
if (obj != null) {
if (keyTime == Long.parseLong(obj.toString())) {
isOpen = true;
return true;
}
if (sleepTime > 0L) {
Thread.sleep(sleepTime);
}
long currentTime = System.currentTimeMillis();
if (currentTime - keyTime > maxWaitTime) {
return false;
}
} else {
ops.setIfAbsent(key, keyTime, lockExTime, TimeUnit.SECONDS);
}
} while (true);
}
/**
* 解锁
*/
public void unlock() {
if (this.lockType != 0) {
throw new RuntimeException("Lock method not supported: unlock");
}
if (isOpen) {
redisTemplate.delete(key);
}
}
/**
* redis等待互斥锁（基于次数的）
*
* @param currentType 互斥类型
* @param lockExTime 锁时长，单位：秒
* @param sleepTime 重试间隔，单位：毫秒
* @param maxWaitTime 最大等待时长，单位：毫秒
* @return 是否获取到锁
* @throws InterruptedException 中断异常
*/
public boolean tryMutexLock(final String currentType, long lockExTime, long sleepTime,
long maxWaitTime) throws InterruptedException {
if (this.lockType != 1) {
throw new RuntimeException("Lock method not supported: tryMutexLock");
}
if (currentType == null || currentType.startsWith("-")) {
throw new RuntimeException("Type is error about MutexLock: " + currentType);
}
ValueOperations<String, Object> ops = redisTemplate.opsForValue();
long keyTime = System.currentTimeMillis();
long lockVersion = timeVersion(keyTime);
long currentLockNum = 1;
String currentStr = currentType + ":" + lockVersion;
String reentryKey = "";
do {
Boolean aBoolean = ops.setIfAbsent(key + ":LOCK", lockVersion, 10, TimeUnit.SECONDS);
if (Boolean.TRUE.equals(aBoolean)) {
String str;
try {
Object obj = ops.get(key);
if (obj == null) {
ops.setIfAbsent(key, currentStr + ":" + 1, lockExTime, TimeUnit.SECONDS);
break;
}
str = obj.toString();
if (str.isEmpty()) {
ops.set(key, currentStr + ":" + 1, lockExTime, TimeUnit.SECONDS);
break;
}
if (str.equals(reentryKey)) {
// 重入
Object objT = ops.get(key);
if (objT != null && objT.equals(reentryKey)) {
ops.set(key, currentStr + ":" + currentLockNum, lockExTime, TimeUnit.SECONDS);
logger.info("检测到重入加入, 重入锁值: {}, 当前锁值: {}:{}", str, currentStr, currentLockNum);
break;
}
continue;
}
} finally {
redisTemplate.delete(key + ":LOCK");
}
if (!str.contains(":")) {
logger.error("RedisLock is ERROR, Format exception: {}", str);
throw new RuntimeException("RedisLock is ERROR, Format exception: " + str);
}
int i = str.indexOf(":");
int j = str.lastIndexOf(":");
String type = str.substring(0, i);
long time = Long.parseLong(str.substring(i + 1, j));
int rNum = Integer.parseInt(str.substring(j + 1));
// 非同类型锁直接等待。同类型新锁时间版本增加，必须再抢锁，但可以快速抢
if (!currentType.equals(type) || lockVersion <= time) {
// 计时器
if (sleepTime > 0L) {
Thread.sleep(sleepTime);
}
currentLockNum = 1;
} else {
reentryKey = str;
currentLockNum = rNum + 1;
// System.out.println("检测到可重入：" + reentryKey);
}
long timeMillis = System.currentTimeMillis();
lockVersion = timeVersion(timeMillis);
if (timeMillis - keyTime > maxWaitTime) {
return false;
}
currentStr = currentType + ":" + lockVersion;
}
} while (true);
// 结束
isOpen = true;
lockTypeVersion = currentType;
lockTimeVersion = lockVersion;
this.lockExTime = lockExTime;
logger.info("创建互斥锁：{}", currentStr);
return true;
}
/**
* 互斥锁解锁（基于次数的）
*/
public void unMutexLock() {
if (this.lockType != 1) {
throw new RuntimeException("Lock method not supported: unMutexLock");
}
if (isOpen) {
ValueOperations<String, Object> ops = redisTemplate.opsForValue();
String unLockStr = "-:0:0";
do {
Boolean aBoolean = ops.setIfAbsent(key + ":LOCK", lockTimeVersion, 10, TimeUnit.SECONDS);
if (Boolean.TRUE.equals(aBoolean)) {
try {
Object obj = ops.get(key);
if (obj == null) {
// 找不到锁了，理论上不会
// 如果存在，锁版本时间相同，且被解锁，但无新入（BUG）
logger.error("检测到解锁意外为空, 待解锁锁值: {}:{}", lockTypeVersion, lockTimeVersion);
return;
}
String str = obj.toString();
if (str.isEmpty()) {
// 找不到锁了，理论上不会，除非值异常
logger.error("检测到解锁空值, 待解锁锁值: {}:{}", lockTypeVersion, lockTimeVersion);
return;
}
if (!str.contains(":")) {
logger.error("RedisLock is ERROR, Format exception: {}", str);
throw new RuntimeException("RedisLock is ERROR, Format exception: " + str);
}
int i = str.indexOf(":");
int j = str.lastIndexOf(":");
String type = str.substring(0, i);
int rNum = Integer.parseInt(str.substring(j + 1));
// 同类型锁才允许释放
if (lockTypeVersion.equals(type)) {
int unLockNum = rNum - 1;
if (unLockNum <= 0) {
redisTemplate.delete(key);
logger.info("解锁成功（解锁）：" + str + " " + unLockStr);
} else {
unLockStr = lockTypeVersion + ":" + lockTimeVersion + ":" + unLockNum;
ops.set(key, unLockStr, lockExTime, TimeUnit.SECONDS);
logger.info("解锁成功（更新锁）：" + str + " " + unLockStr);
}
return;
} else {
// 锁已经被意外更新，理论上不会
// 如果存在，锁版本时间相同，且被解锁，但有新类型插入（可能是BUG）
logger.warn("检测到解锁意外更新, 当前锁值: {}, 待解锁锁值: {}:{}",
str, lockTypeVersion, lockTimeVersion);
return;
}
} finally {
redisTemplate.delete(key + ":LOCK");
}
}
} while (true);
}
}
private static long timeVersion(long timeMillis) {
long nanoTime = System.nanoTime();
return (timeMillis * 100_000L + (nanoTime % 10_000_000L) / 100);
}
/**
* redis等待互斥锁（基于时间的，只要求时间序列的）
*
* @param currentType 互斥类型
* @param lockExTime 锁时长，单位：秒
* @param sleepTime 重试间隔，单位：毫秒
* @param maxWaitTime 最大等待时长，单位：毫秒
* @return 是否获取到锁
* @throws InterruptedException 中断异常
*/
public boolean tryMutexTimeLock(final String currentType, long lockExTime, long sleepTime,
long maxWaitTime) throws InterruptedException {
if (this.lockType != 2) {
throw new RuntimeException("Lock method not supported: tryMutexTimeLock");
}
if (currentType == null || currentType.startsWith("-")) {
throw new RuntimeException("Type is error about MutexLock: " + currentType);
}
ValueOperations<String, Object> ops = redisTemplate.opsForValue();
long keyTime = System.currentTimeMillis();
long currentTime = timeVersion(keyTime);
String currentStr = currentType + ":" + currentTime;
String reentryKey = "";
do {
Object obj = ops.get(key);
if (obj == null) {
ops.setIfAbsent(key, currentStr, lockExTime, TimeUnit.SECONDS);
continue;
}
String str = obj.toString();
if (str.isEmpty() || str.equals(reentryKey)) {
// 重入
Boolean aBoolean = ops.setIfAbsent(key + ":LOCK", currentTime, 10, TimeUnit.SECONDS);
if (Boolean.TRUE.equals(aBoolean)) {
try {
ops.set(key, currentStr, lockExTime, TimeUnit.SECONDS);
// System.out.println("检测到重入加入：" + reentryKey);
} finally {
redisTemplate.delete(key + ":LOCK");
}
}
continue;
}
// 结束
if (str.equals(currentStr)) {
isOpen = true;
lockTypeVersion = currentType;
lockTimeVersion = currentTime;
// System.out.println("创建锁：" + lockTypeVersion + ":" + lockTimeVersion);
return true;
}
if (!str.contains(":")) {
logger.error("RedisLock is ERROR, Format exception: {}", str);
throw new RuntimeException("RedisLock is ERROR, Format exception: " + str);
}
int i = str.indexOf(":");
String type = str.substring(0, i);
long time = Long.parseLong(str.substring(i + 1));
// 非同类型锁直接等待。同类型新锁时间版本增加，必须再抢锁，但可以快速抢
if (!currentType.equals(type) || currentTime <= time) {
// 计时器
if (sleepTime > 0L) {
Thread.sleep(sleepTime);
}
} else {
reentryKey = str;
// System.out.println("检测到可重入：" + reentryKey);
}
long timeMillis = System.currentTimeMillis();
currentTime = timeVersion(timeMillis);
if (timeMillis - keyTime > maxWaitTime) {
return false;
}
currentStr = currentType + ":" + currentTime;
} while (true);
}
/**
* 互斥锁解锁（基于时间的，只要求时间序列的）
*/
public void unMutexTimeLock() {
if (this.lockType != 2) {
throw new RuntimeException("Lock method not supported: unMutexTimeLock");
}
if (isOpen) {
ValueOperations<String, Object> ops = redisTemplate.opsForValue();
String unLockStr = "-:0";
do {
Object obj = ops.get(key);
if (obj == null) {
// 找不到锁了，理论上不会
// 如果存在，锁版本时间相同，且被解锁，但无新入（时间BUG）
logger.error("检测到解锁意外为空, 待解锁锁值: {}:{}", lockTypeVersion, lockTimeVersion);
break;
}
String str = obj.toString();
if (str.isEmpty()) {
// 找不到锁了，理论上不会，除非值异常
logger.error("检测到解锁空值, 待解锁锁值: {}:{}", lockTypeVersion, lockTimeVersion);
break;
}
if (str.equals(unLockStr)) {
// 抢占成功，解锁
Boolean aBoolean = ops.setIfAbsent(key + ":LOCK", lockTimeVersion, 10, TimeUnit.SECONDS);
if (Boolean.TRUE.equals(aBoolean)) {
try {
redisTemplate.delete(key);
// System.out.println("抢占成功解锁：" + str + " " + unLockStr);
} finally {
redisTemplate.delete(key + ":LOCK");
}
break;
}
}
if (!str.contains(":")) {
logger.error("RedisLock is ERROR, Format exception: {}", str);
throw new RuntimeException("RedisLock is ERROR, Format exception: " + str);
}
int i = str.indexOf(":");
String type = str.substring(0, i);
long time = Long.parseLong(str.substring(i + 1));
// 同类型锁才允许释放
if (lockTypeVersion.equals(type) || ("-" + lockTypeVersion).equals(type)) {
if (lockTimeVersion < time) {
// 锁版本时间小于现有的锁, 不做处理
// System.out.println("检测到解锁时间小于：" + str + ", " + time + " > " + lockTimeVersion);
break;
} else {
if (lockTimeVersion > time) {
// 锁版本时间小于现有的锁, 不做处理
// System.out.println("检测到解锁时间大于：" + str + ", " + time + " > " + lockTimeVersion);
}
// 完全相同或大于，则抢占解锁
// 此时只有同类型的会更新时间版本，但别的都进不了
Boolean aBoolean = ops.setIfAbsent(key + ":LOCK", lockTimeVersion, 10, TimeUnit.SECONDS);
if (Boolean.TRUE.equals(aBoolean)) {
try {
unLockStr = "-" + lockTypeVersion + ":" + lockTimeVersion;
ops.set(key, unLockStr, 10, TimeUnit.SECONDS);
// System.out.println("抢占解锁：" + str + " " + unLockStr);
} finally {
redisTemplate.delete(key + ":LOCK");
}
}
}
} else {
// 锁已经被意外更新，理论上不会
// 如果存在，锁版本时间相同，且被解锁，但有新类型插入（可能是BUG）
logger.warn("检测到解锁意外更新, 当前锁值: {}, 待解锁锁值: {}:{}",
str, lockTypeVersion, lockTimeVersion);
break;
}
} while (true);
}
}
}
