当前位置：首页 > news >正文

Java-函数式编程-实现分布式锁工具

news 2026/5/12 21:24:26

该业务适用场景:

分布式环境下的资源互斥访问
防止重复执行（如定时任务）
库存扣减等需要强一致性的操作

一. 背景

在开发过程中需要使用到分布式锁的时候(以redisson为例). 一般会通过初始化RedissonClient配置, 然后在需要的地方使用. 当使用的多的时候会发现, 代码中充斥着相似的代码结构, 例如

String lockKey = RedisBaseKeyEnum.APP_USER_REGISTRY_VERIFY_CODE_KEY_LOCK.buildKey(account);
RLock lock = redissonClient.getLock(lockKey);
try {if (lock != null && !lock.isLocked() && lock.tryLock(0, 60L, TimeUnit.SECONDS)) {// some biz}
} catch (InterruptedException e) {LOG.error("加锁失败，key:{}, account：{}", lockKey, account, e);Thread.currentThread().interrupt();throw new BizException(BizErrorCode.VERIFY_CODE_SENDER_ERROR);
} finally {if (lock.isLocked() && lock.isHeldByCurrentThread()) {lock.unlock();}
}

简单的加锁过程类似于上面的代码. 所以为了提升代码的可复用性, 降低重复代码, 提高系统可维护性. 我们可以通过Java中的函数式编程, 将重复的代码块抽象出一个模板. 然后具体的业务通过作为方法入参 来实现.

二. 核心点

抽象分布式锁业务
使用@FunctionalInterface注解将需要注入的业务抽象出来
注入使用

三. 实现方式

定义函数式接口 @FunctionalInterface

============================= 加锁业务 有返回值 ===============================
package com.sj.utils.lock.func;@FunctionalInterface
public interface LockTask {void run() throws Exception; // 没有入参的方法
}
============================= 加锁业务没有返回值 ===============================
package com.sj.utils.lock.func;@FunctionalInterface
public interface LockTask {void run() throws Exception; // 没有入参的方法}

定义Redisson分布式锁类型枚举

package com.sj.utils.lock;public enum RLockType {REENTRANT,       // 可重入锁FAIR,            // 公平锁READ,            // 读锁WRITE,           // 写锁TRY,             // 尝试获取锁TRY_WITH_FAIL    // 尝试失败锁 tryLock中waitTime为0
}

然后

===========================RLockSimpleUtil========================
public RLock getLock(String lockKey, RLockType lockType) {switch (lockType) {case TRY:case REENTRANT:case TRY_WITH_FAIL:return redissonClient.getLock(lockKey);case FAIR:return redissonClient.getFairLock(lockKey);case READ:return redissonClient.getReadWriteLock(lockKey).readLock();case WRITE:return redissonClient.getReadWriteLock(lockKey).writeLock();default:throw new IllegalArgumentException("Unknown lock type: " + lockType);}
}

定义RLockSimpleUtil

package com.sj.utils.lock;import com.sj.utils.lock.func.LockException;
import com.sj.utils.lock.func.LockTask;
import com.sj.utils.lock.func.LockTaskWithResult;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;import java.util.concurrent.TimeUnit;@Component
public class RLockSimpleUtil {private final Logger log = LoggerFactory.getLogger(RLockSimpleUtil.class);private final RedissonClient redissonClient;// 锁默认参数private final long DEFAULT_WAIT_TIME = 5;       // 秒private final long DEFAULT_LEASE_TIME = 30;     // 秒private final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.SECONDS;public RLockSimpleUtil(RedissonClient redissonClient) {this.redissonClient = redissonClient;}/*** 执行带返回值的任务，默认使用可重入锁** @param lockKey 锁键* @param task    任务* @param <T>     任务返回值类型* @return 任务返回值*/public <T> T executeWithResult(String lockKey, LockTaskWithResult<T> task) {return executeWithResult(lockKey, RLockType.REENTRANT, DEFAULT_WAIT_TIME, DEFAULT_LEASE_TIME, DEFAULT_TIME_UNIT, task);}/*** 执行带返回值的任务，支持指定锁类型和超时时间** @param lockKey   锁键* @param lockType  锁类型* @param waitTime  等待时间* @param leaseTime 租约时间* @param unit      时间单位* @param lockTask  任务* @param <T>       任务返回值类型* @return 任务返回值*/public <T> T executeWithResult(String lockKey,RLockType lockType,long waitTime,long leaseTime,TimeUnit unit,LockTaskWithResult<T> lockTask) {RLock lock = getLock(lockKey, lockType);boolean acquired = false;try {acquired = lock.tryLock(waitTime, leaseTime, unit);if (!acquired) {throw new LockException("Failed to acquire lock: " + lockKey);}return lockTask.runWithResult();} catch (InterruptedException e) {Thread.currentThread().interrupt();throw new LockException("Lock acquisition interrupted", e);} catch (Exception e) {throw new RuntimeException(e);} finally {unlockQuietly(lock, acquired);}}/*** 执行无返回值的任务，默认使用可重入锁** @param lockKey  锁键* @param lockTask 任务*/public void execute(String lockKey, LockTask lockTask) {execute(lockKey, RLockType.REENTRANT, DEFAULT_WAIT_TIME, DEFAULT_LEASE_TIME, DEFAULT_TIME_UNIT, lockTask);}/*** 执行无返回值的任务，支持指定锁类型和超时时间** @param lockKey   锁键* @param lockType  锁类型* @param waitTime  等待时间* @param leaseTime 租约时间* @param timeUnit  时间单位* @param lockTask  任务*/public void execute(String lockKey, RLockType lockType, long waitTime, long leaseTime, TimeUnit timeUnit, LockTask lockTask) {RLock lock = getLock(lockKey, lockType);boolean acquired = false;try {acquired = lock.tryLock(waitTime, leaseTime, timeUnit);if (!acquired) {throw new LockException("Failed to acquire lock: " + lockKey);}lockTask.run();} catch (InterruptedException e) {Thread.currentThread().interrupt();throw new LockException("Interrupted while trying to acquire lock: " + lockKey);} catch (Exception e) {throw new RuntimeException(e);} finally {unlockQuietly(lock, acquired);}}/*** 获取指定锁类型的锁实例** @param lockKey  锁键* @param lockType 锁类型* @return 锁实例*/public RLock getLock(String lockKey, RLockType lockType) {switch (lockType) {case TRY:case REENTRANT:case TRY_WITH_FAIL:return redissonClient.getLock(lockKey);case FAIR:return redissonClient.getFairLock(lockKey);case READ:return redissonClient.getReadWriteLock(lockKey).readLock();case WRITE:return redissonClient.getReadWriteLock(lockKey).writeLock();default:throw new IllegalArgumentException("Unknown lock type: " + lockType);}}/*** 安静地解锁锁实例** @param lock     锁实例* @param acquired 是否成功获取锁*/private void unlockQuietly(RLock lock, boolean acquired) {try {if (acquired && lock.isLocked() && lock.isHeldByCurrentThread()) {lock.unlock();}} catch (Exception e) {if (lock != null && lock.isHeldByCurrentThread()) {log.warn("Error unlocking: {} ", e.getMessage(), e);lock.unlock();}}}}

四. 使用方式

在具体的业务中注入RLockSimpleUtil
根据具体业务, 直接调用util中的具体方法, 例如一个简单的加锁业务:

@Override
public ShortAddressResult acquire(ShortAddressTypeEnum type, Long householdId, Integer count) {String key = RedisBaseKeyEnum.DEVICE_SHORT_ADDRESS_LIST_CACHE_KEY_LOCK.buildKey(householdId.toString());return rLockSimpleUtil.executeWithResult(key,RLockType.FAIR,30,3,TimeUnit.SECONDS,// 下方执行具体的加锁业务() -> {String cacheKey = getCacheKey(type, householdId);return shortAddressPoolService.acquire(type, cacheKey, householdId, count);});
}