如果大家項目中Redis是多機部署的可以來好好看看這篇實現,講的非常好。
使用Redisson實現分布式鎖,Spring AOP簡化之
Redisson概述
Redisson是一個在Redis的基礎上實現的Java駐內存數據網格(In-Memory Data Grid)。它不僅提供了一系列的分布式的Java常用對象,還提供了許多分布式服務。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最簡單和最便捷的方法。Redisson的宗旨是促進使用者對Redis的關注分離(Separation of Concern),從而讓使用者能夠將精力更集中地放在處理業務邏輯上。
Redisson底層采用的是Netty 框架。支持Redis 2.8以上版本,支持Java1.6+以上版本。
關於Redisson更多詳細介紹,可參考Redssion概述
Redisson提供的分布式鎖
可重入鎖
Redisson的分布式可重入鎖RLock Java對象實現了java.util.concurrent.locks.Lock接口,同時還支持自動過期解鎖。下面是RLock的基本使用方法:
RLock lock = redisson.getLock("anyLock"); // 最常見的使用方法 lock.lock(); // 支持過期解鎖功能 // 10秒鍾以后自動解鎖 // 無需調用unlock方法手動解鎖 lock.lock(10, TimeUnit.SECONDS); // 嘗試加鎖,最多等待100秒,上鎖以后10秒自動解鎖 boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS); ... lock.unlock();
Redisson同時還為分布式鎖提供了異步執行的相關方法:
RLock lock = redisson.getLock("anyLock"); lock.lockAsync(); lock.lockAsync(10, TimeUnit.SECONDS); Future<Boolean> res = lock.tryLockAsync(100, 10, TimeUnit.SECONDS);
公平鎖
Redisson分布式可重入公平鎖也是實現了java.util.concurrent.locks.Lock接口的一種RLock對象。在提供了自動過期解鎖功能的同時,保證了當多個Redisson客戶端線程同時請求加鎖時,優先分配給先發出請求的線程。
RLock fairLock = redisson.getFairLock("anyLock"); // 最常見的使用方法 fairLock.lock(); // 支持過期解鎖功能 // 10秒鍾以后自動解鎖 // 無需調用unlock方法手動解鎖 fairLock.lock(10, TimeUnit.SECONDS); // 嘗試加鎖,最多等待100秒,上鎖以后10秒自動解鎖 boolean res = fairLock.tryLock(100, 10, TimeUnit.SECONDS); ... fairLock.unlock();
其他鎖
Redisson還提供了其他機制的鎖,如聯鎖(MultiLock)、紅鎖(RedLock)等。詳細可參考:分布式鎖和同步器
使用Redisson實現分布式鎖
- 定義回調接口
/** * 分布式鎖回調接口 */ public interface DistributedLockCallback<T> { /** * 調用者必須在此方法中實現需要加分布式鎖的業務邏輯 * * @return */ public T process(); /** * 得到分布式鎖名稱 * * @return */ public String getLockName(); }
- 定義分布式鎖模板
/** * 分布式鎖操作模板 */ public interface DistributedLockTemplate { long DEFAULT_WAIT_TIME = 30; long DEFAULT_TIMEOUT = 5; TimeUnit DEFAULT_TIME_UNIT = TimeUnit.SECONDS; /** * 使用分布式鎖,使用鎖默認超時時間。 * @param callback * @param fairLock 是否使用公平鎖 * @return */ <T> T lock(DistributedLockCallback<T> callback, boolean fairLock); /** * 使用分布式鎖。自定義鎖的超時時間 * * @param callback * @param leaseTime 鎖超時時間。超時后自動釋放鎖。 * @param timeUnit * @param fairLock 是否使用公平鎖 * @param <T> * @return */ <T> T lock(DistributedLockCallback<T> callback, long leaseTime, TimeUnit timeUnit, boolean fairLock); /** * 嘗試分布式鎖,使用鎖默認等待時間、超時時間。 * @param callback * @param fairLock 是否使用公平鎖 * @param <T> * @return */ <T> T tryLock(DistributedLockCallback<T> callback, boolean fairLock); /** * 嘗試分布式鎖,自定義等待時間、超時時間。 * @param callback * @param waitTime 獲取鎖最長等待時間 * @param leaseTime 鎖超時時間。超時后自動釋放鎖。 * @param timeUnit * @param fairLock 是否使用公平鎖 * @param <T> * @return */ <T> T tryLock(DistributedLockCallback<T> callback, long waitTime, long leaseTime, TimeUnit timeUnit, boolean fairLock); }
- 實現分布式鎖模板
public class SingleDistributedLockTemplate implements DistributedLockTemplate { private RedissonClient redisson; public SingleDistributedLockTemplate() { } public SingleDistributedLockTemplate(RedissonClient redisson) { this.redisson = redisson; } @Override public <T> T lock(DistributedLockCallback<T> callback, boolean fairLock) { return lock(callback, DEFAULT_TIMEOUT, DEFAULT_TIME_UNIT, fairLock); } @Override public <T> T lock(DistributedLockCallback<T> callback, long leaseTime, TimeUnit timeUnit, boolean fairLock) { RLock lock = getLock(callback.getLockName(), fairLock); try { lock.lock(leaseTime, timeUnit); return callback.process(); } finally { if (lock != null && lock.isLocked()) { lock.unlock(); } } } @Override public <T> T tryLock(DistributedLockCallback<T> callback, boolean fairLock) { return tryLock(callback, DEFAULT_WAIT_TIME, DEFAULT_TIMEOUT, DEFAULT_TIME_UNIT, fairLock); } @Override public <T> T tryLock(DistributedLockCallback<T> callback, long waitTime, long leaseTime, TimeUnit timeUnit, boolean fairLock) { RLock lock = getLock(callback.getLockName(), fairLock); try { if (lock.tryLock(waitTime, leaseTime, timeUnit)) { return callback.process(); } } catch (InterruptedException e) { } finally { if (lock != null && lock.isLocked()) { lock.unlock(); } } return null; } private RLock getLock(String lockName, boolean fairLock) { RLock lock; if (fairLock) { lock = redisson.getFairLock(lockName); } else { lock = redisson.getLock(lockName); } return lock; } public void setRedisson(RedissonClient redisson) { this.redisson = redisson; } }
- 使用SingleDistributedLockTemplate
DistributedLockTemplate lockTemplate = ...;
final String lockName = ...; lockTemplate.lock(new DistributedLockCallback<Object>() { @Override public Object process() { //do some business return null; } @Override public String getLockName() { return lockName; } }, false);
但是每次使用分布式鎖都要寫類似上面的重復代碼,有沒有什么方法可以只關注核心業務邏輯代碼的編寫,即上面的"do some business"。下面介紹如何使用Spring AOP來實現這一目標。
使用Spring AOP簡化分布式鎖
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface DistributedLock { /** * 鎖的名稱。 * 如果lockName可以確定,直接設置該屬性。 */ String lockName() default ""; /** * lockName后綴 */ String lockNamePre() default ""; /** * lockName后綴 */ String lockNamePost() default "lock"; /** * 獲得鎖名時拼接前后綴用到的分隔符 * @return */ String separator() default "."; /** * <pre> * 獲取注解的方法參數列表的某個參數對象的某個屬性值來作為lockName。因為有時候lockName是不固定的。 * 當param不為空時,可以通過argNum參數來設置具體是參數列表的第幾個參數,不設置則默認取第一個。 * </pre> */ String param() default ""; /** * 將方法第argNum個參數作為鎖 */ int argNum() default 0; /** * 是否使用公平鎖。 * 公平鎖即先來先得。 */ boolean fairLock() default false; /** * 是否使用嘗試鎖。 */ boolean tryLock() default false; /** * 最長等待時間。 * 該字段只有當tryLock()返回true才有效。 */ long waitTime() default 30L; /** * 鎖超時時間。 * 超時時間過后,鎖自動釋放。 * 建議: * 盡量縮簡需要加鎖的邏輯。 */ long leaseTime() default 5L; /** * 時間單位。默認為秒。 */ TimeUnit timeUnit() default TimeUnit.SECONDS; }
- 定義切面織入的代碼
@Aspect @Component public class DistributedLockAspect { @Autowired private DistributedLockTemplate lockTemplate; @Pointcut("@annotation(cn.sprinkle.study.distributedlock.common.annotation.DistributedLock)") public void DistributedLockAspect() {} @Around(value = "DistributedLockAspect()") public Object doAround(ProceedingJoinPoint pjp) throws Throwable { //切點所在的類 Class targetClass = pjp.getTarget().getClass(); //使用了注解的方法 String methodName = pjp.getSignature().getName(); Class[] parameterTypes = ((MethodSignature)pjp.getSignature()).getMethod().getParameterTypes(); Method method = targetClass.getMethod(methodName, parameterTypes); Object[] arguments = pjp.getArgs(); final String lockName = getLockName(method, arguments); return lock(pjp, method, lockName); } @AfterThrowing(value = "DistributedLockAspect()", throwing="ex") public void afterThrowing(Throwable ex) { throw new RuntimeException(ex); } public String getLockName(Method method, Object[] args) { Objects.requireNonNull(method); DistributedLock annotation = method.getAnnotation(DistributedLock.class); String lockName = annotation.lockName(), param = annotation.param(); if (isEmpty(lockName)) { if (args.length > 0) { if (isNotEmpty(param)) { Object arg; if (annotation.argNum() > 0) { arg = args[annotation.argNum() - 1]; } else { arg = args[0]; } lockName = String.valueOf(getParam(arg, param)); } else if (annotation.argNum() > 0) { lockName = args[annotation.argNum() - 1].toString(); } } } if (isNotEmpty(lockName)) { String preLockName = annotation.lockNamePre(), postLockName = annotation.lockNamePost(), separator = annotation.separator(); StringBuilder lName = new StringBuilder(); if (isNotEmpty(preLockName)) { lName.append(preLockName).append(separator); } lName.append(lockName); if (isNotEmpty(postLockName)) { lName.append(separator).append(postLockName); } lockName = lName.toString(); return lockName; } throw new IllegalArgumentException("Can't get or generate lockName accurately!"); } /** * 從方法參數獲取數據 * * @param param * @param arg 方法的參數數組 * @return */ public Object getParam(Object arg, String param) { if (isNotEmpty(param) && arg != null) { try { Object result = PropertyUtils.getProperty(arg, param); return result; } catch (NoSuchMethodException e) { throw new IllegalArgumentException(arg + "沒有屬性" + param + "或未實現get方法。", e); } catch (Exception e) { throw new RuntimeException("", e); } } return null; } public Object lock(ProceedingJoinPoint pjp, Method method, final String lockName) { DistributedLock annotation = method.getAnnotation(DistributedLock.class); boolean fairLock = annotation.fairLock(); boolean tryLock = annotation.tryLock(); if (tryLock) { return tryLock(pjp, annotation, lockName, fairLock); } else { return lock(pjp,lockName, fairLock); } } public Object lock(ProceedingJoinPoint pjp, final String lockName, boolean fairLock) { return lockTemplate.lock(new DistributedLockCallback<Object>() { @Override public Object process() { return proceed(pjp); } @Override public String getLockName() { return lockName; } }, fairLock); } public Object tryLock(ProceedingJoinPoint pjp, DistributedLock annotation, final String lockName, boolean fairLock) { long waitTime = annotation.waitTime(), leaseTime = annotation.leaseTime(); TimeUnit timeUnit = annotation.timeUnit(); return lockTemplate.tryLock(new DistributedLockCallback<Object>() { @Override public Object process() { return proceed(pjp); } @Override public String getLockName() { return lockName; } }, waitTime, leaseTime, timeUnit, fairLock); } public Object proceed(ProceedingJoinPoint pjp) { try { return pjp.proceed(); } catch (Throwable throwable) { throw new RuntimeException(throwable); } } private boolean isEmpty(Object str) { return str == null || "".equals(str); } private boolean isNotEmpty(Object str) { return !isEmpty(str); } }
有了上面兩段代碼,以后需要用到分布式鎖,只需在核心業務邏輯方法添加注解@DistributedLock,並設置LockName、fairLock等即可。下面的DistributionService演示了多種使用情景。
@Service public class DistributionService { @Autowired private RedissonClient redissonClient; @DistributedLock(param = "id", lockNamePost = ".lock") public Integer aspect(Person person) { RMap<String, Integer> map = redissonClient.getMap("distributionTest"); Integer count = map.get("count"); if (count > 0) { count = count - 1; map.put("count", count); } return count; } @DistributedLock(argNum = 1, lockNamePost = ".lock") public Integer aspect(String i) { RMap<String, Integer> map = redissonClient.getMap("distributionTest"); Integer count = map.get("count"); if (count > 0) { count = count - 1; map.put("count", count); } return count; } @DistributedLock(lockName = "lock", lockNamePost = ".lock") public int aspect(Action<Integer> action) { return action.action(); } }
- 測試
定義一個Worker類:
public class Worker implements Runnable { private final CountDownLatch startSignal; private final CountDownLatch doneSignal; private final DistributionService service; private RedissonClient redissonClient; public Worker(CountDownLatch startSignal, CountDownLatch doneSignal, DistributionService service, RedissonClient redissonClient) { this.startSignal = startSignal; this.doneSignal = doneSignal; this.service = service; this.redissonClient = redissonClient; } @Override public void run() { try { startSignal.await(); System.out.println(Thread.currentThread().getName() + " start"); // Integer count = service.aspect(new Person(1, "張三")); // Integer count = service.aspect("1"); Integer count = service.aspect(() -> { RMap<String, Integer> map = redissonClient.getMap("distributionTest"); Integer count1 = map.get("count"); if (count1 > 0) { count1 = count1 - 1; map.put("count", count1); } return count1; }); System.out.println(Thread.currentThread().getName() + ": count = " + count); doneSignal.countDown(); } catch (InterruptedException ex) { System.out.println(ex); } } }
定義Controller類:
@RestController @RequestMapping("/distributedLockTest") public class DistributedLockTestController { private int count = 10; @Autowired private RedissonClient redissonClient; @Autowired private DistributionService service; @RequestMapping(method = RequestMethod.GET) public String distributedLockTest() throws Exception { RMap<String, Integer> map = redissonClient.getMap("distributionTest"); map.put("count", 8); CountDownLatch startSignal = new CountDownLatch(1); CountDownLatch doneSignal = new CountDownLatch(count); for (int i = 0; i < count; ++i) { // create and start threads new Thread(new Worker(startSignal, doneSignal, service)).start(); } startSignal.countDown(); // let all threads proceed doneSignal.await(); System.out.println("All processors done. Shutdown connection"); return "finish"; } }
Redisson基本配置:
singleServerConfig: idleConnectionTimeout: 10000 pingTimeout: 1000 connectTimeout: 10000 timeout: 3000 retryAttempts: 3 retryInterval: 1500 reconnectionTimeout: 3000 failedAttempts: 3 password: subscriptionsPerConnection: 5 clientName: null address: "redis://127.0.0.1:6379" subscriptionConnectionMinimumIdleSize: 1 subscriptionConnectionPoolSize: 50 connectionMinimumIdleSize: 10 connectionPoolSize: 64 database: 0 dnsMonitoring: false dnsMonitoringInterval: 5000 threads: 0 nettyThreads: 0 codec: !<org.redisson.codec.JsonJacksonCodec> {} useLinuxNativeEpoll: false
工程中需要注入的對象:
@Value("classpath:/redisson-conf.yml") Resource configFile; @Bean(destroyMethod = "shutdown") RedissonClient redisson() throws IOException { Config config = Config.fromYAML(configFile.getInputStream()); return Redisson.create(config); } @Bean DistributedLockTemplate distributedLockTemplate(RedissonClient redissonClient) { return new SingleDistributedLockTemplate(redissonClient); }
需要引入的依賴:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-aop</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.5.3</version> </dependency> <dependency> <groupId>commons-beanutils</groupId> <artifactId>commons-beanutils</artifactId> <version>1.8.3</version> </dependency>
最后啟動工程,然后訪問localhost:8080/distributedLockTest,可以看到如下結果:
觀察結果,可以看出,10個線程中只有8個線程能執行count減1操作,而且多個線程是依次執行的。也就是說分布式鎖起作用了。
使用lambda
該注解還可以配合lambda使用。在介紹之前,先科普一下使用spring注解時需要注意的地方,有兩點。
第一,在使用spring提供的方法注解時,比較常用的是@Transactional
注解。若是Service層不帶注解的方法A調用同一個Service類帶@Transactional
注解的方法B,那么方法B的事務注解將不起作用。比如:
...
public void methodA() { methodB(); } @Transactional public void methodB() { // 操作表A // 操作表B } ...
上面的代碼中,假設有一次調用方法A,方法A又調用方法B,但是此次調用在操作表B時出錯了。我們的意願是這樣的:之前對表A的操作回滾。但實際上卻不會回滾,因為此時的@Transactional
注解並不會生效。原因是調用方法B是同一個Service的方法A,而若是在其他類中調用方法B注解才生效。這也就不難解釋為什么注解加在private方法上是不起作用的了。因為private方法只能在同一個方法中調用。
上面所說的調用同一個類的帶注解的方法,該注解將不生效,感興趣的可以自己找找原因,這里就不細說了。
第二,注解(包括spring提供的、自定義的)加在普通類的方法上,spring是掃描不到的。普通類指類簽名上沒有諸如@Service
等Spring提供的注解(因為此分布式鎖集成使用的是spring aop,所以介紹的都是與spring相關的)。比如,如果把上面貼出的DistributionService
中的各個方法放到Worker
中,那么這些注解將不起作用,因為Worker
類簽名並沒有加任何注解,所以spring在掃描的時候直接跳過該類,因此定義在Worker
中的帶@DistributedLock
注解的方法(如果有的話)也就無法被掃描到。
在上面貼出的代碼中,Worker中需要使用分布式鎖的業務邏輯比較簡單,所以都寫到DistributionService
中,但在實際開發中,我們通常有把業務邏輯直接寫在Worker
中的需求,畢竟是與Worker
相關的,放到哪一個Service都感覺很別扭。所以,我們可以定義一個分布式鎖管理器,如DistributedLockManager
,然后在初始化Worker時引入即可。接下來改造Worker
和定義DistributedLockManager
:
Worker1:
public class Worker1 implements Runnable { private final CountDownLatch startSignal; private final CountDownLatch doneSignal; private final DistributedLockManager distributedLockManager; private RedissonClient redissonClient; public Worker1(CountDownLatch startSignal, CountDownLatch doneSignal, DistributedLockManager distributedLockManager, RedissonClient redissonClient) { this.startSignal = startSignal; this.doneSignal = doneSignal; this.distributedLockManager = distributedLockManager; this.redissonClient = redissonClient; } @Override public void run() { try { System.out.println(Thread.currentThread().getName() + " start"); startSignal.await(); Integer count = aspect("lock"); System.out.println(Thread.currentThread().getName() + ": count = " + count); doneSignal.countDown(); } catch (Exception e) { e.printStackTrace(); } } public int aspect(String lockName) { return distributedLockManager.aspect(lockName, this); } public int aspectBusiness(String lockName) { RMap<String, Integer> map = redissonClient.getMap("distributionTest"); Integer count = map.get("count"); if (count > 0) { count = count - 1; map.put("count", count); } return count; } }
DistributedLockManager:
@Component public class DistributedLockManager { @DistributedLock(argNum = 1, lockNamePost = ".lock") public Integer aspect(String lockName, Worker1 worker1) { return worker1.aspectBusiness(lockName); } }
這樣做,雖然可以將業務從Service層抽離出來,放到分布式鎖管理器DistributedLockManager
統一管理,但每次都需要將Worker
一起傳過去,同樣感覺很別扭。那么有沒有更好的辦法呢?有,使用lambda。(上面鋪墊了那么多,終於進入正題了!o(╥﹏╥)o)
lambda是java 8的新特性之一,若未了解過的建議先去惡補一番。因為java 8支持lambda,所以也新加了很多函數式接口,這里簡單列幾個:
函數式接口 | 參數類型 | 返回類型 | 描述 |
---|---|---|---|
Supplier<T> | 無 | T | 提供一個T類型的值 |
Consumer<T> | T | void | 處理一個T類型的值 |
BiConsumer<T, U> | T, U | void | 處理T類型和U類型的值 |
Predicate<T> | T | boolean | 一個 計算Boolean值的函數 |
Function<T, R> | T | R | 一個參數類型為T的函數 |
ToIntFunction<T> ToLongFunction<T> ToDoubleFunction<T> |
T | int long double |
分別計算int、long、double值的函數 |
IntFunction<R> LongFunction<R> DoubleFunction<R> |
int long double |
R | 參數分別為int、long、double類型的函數 |
BiFunction<T, U, R> | T, U | R | 一個參數類型為T和U的函數 |
UnaryOperator<T> | T | T | 對類型T進行的一元操作 |
BinaryOperator<T> | T, T | T | 對類型T進行的二元操作 |
觀察Worker1
中方法aspect(Person)
的邏輯,最后需要返回一個int值,所以我們可以使用Supplier<T>
來作為參數的類型,在分布式鎖管理器中添加一個方法,如下:
@DistributedLock(lockName = "lock", lockNamePost = ".lock") public int aspect(Supplier<Integer> supplier) { return supplier.get(); }
然后,在Worker1
中也定義一個方法:
private int aspect() { RMap<String, Integer> map = redissonClient.getMap("distributionTest"); Integer count1 = map.get("count"); if (count1 > 0) { count1 = count1 - 1; map.put("count", count1); } return count1; }
最后在Worker1
的run方法中使用,把Integer count = aspect("lock");
替換成如下:
Integer count = distributedLockManager.aspect(() -> { return aspect(); });
其實也可以簡寫:
Integer count = distributedLockManager.aspect(() -> aspect());
通過這樣改造,是不是發現優雅多了。
測試
在DistributedLockTestController
中,幫下面的代碼替換成另一段代碼:
for (int i = 0; i < count; ++i) { // create and start threads new Thread(new Worker(startSignal, doneSignal, service, redissonClient)).start(); }
替換成:
for (int i = 0; i < count; ++i) { // create and start threads new Thread(new Worker1(startSignal, doneSignal, distributedLockManager, redissonClient)).start(); }
最后啟動工程,訪問http://localhost:8080/distributedLockTest,可以看到類似如下的結果:
另外,因為暫時沒有找到合適的參數類型為“無”、返回類型也為“無”的函數式接口(找到一個——Runnable,但如果用了,怕產生歧義,所以就算了),既然如此,我們不妨自己定義一個,如下:
@FunctionalInterface public interface Action { void action(); }
使用也很簡單,在DistributedLockManager
定義類似如下的方法:
@DistributedLock(lockName = "lock", lockNamePost = ".lock") public void doSomething(Action action) { action.action(); }
然后,在需要的地方這樣用:
distributedLockManager.doSomething(() -> { // do something });
至此,使用Redisson實現分布式鎖,然后使用Spring AOP簡化分布式鎖介紹完畢。
若有什么地方有錯誤的或需要改進的,歡迎留言一起討論交流。