一、場景

　　項目A監聽mq中的其他項目的部署消息（包括push_seq, status, environment，timestamp等），然后將部署消息同步到數據庫中（項目X在對應環境[environment]上部署的push_seq[項目X的版本]）。那么問題來了，mq中加入包含了兩個部署消息 dm1 和 dm2，dm2的push_seq > dm1的push_seq，在分布式的情況下，dm1 和 dm2可能會分別被消費（也就是并行），那么在同步數據庫的時候可能會發生 dm1 的數據保存 后于 dm2的數據保存，導致保存項目的部署信息發生異常。

二、解決思路

　　將mq消息的并行消費變成串行消費，這里借助redis分布式鎖來完成。同一個服務在分布式的狀態下，監聽到mq消息后，觸發方法的執行，執行之前（通過spring aop around來做的）首先獲得redis的一個分布式鎖，獲取鎖成功之后才能執行相關的邏輯以及數據庫的保存，最后釋放鎖。

三、主要代碼

import java.lang.annotation.ElementType; import java.lang.annotation.Inherited; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target;/** * @author: hujunzheng * @create: 17/9/29 下午2:49 */ @Target({ElementType.METHOD}) @Retention(RetentionPolicy.RUNTIME) @Inherited public @interface RedisLock {/*** redis的key* @return*/String value();/*** 持鎖時間,單位毫秒,默認一分鐘*/long keepMills() default 60000;/*** 當獲取失敗時候動作*/LockFailAction action() default LockFailAction.GIVEUP;public enum LockFailAction{/*** 放棄*/GIVEUP,/*** 繼續*/CONTINUE;}/*** 睡眠時間,設置GIVEUP忽略此項* @return*/long sleepMills() default 500; }

?

import java.lang.reflect.Method; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.annotation.Around; import org.aspectj.lang.annotation.Aspect; import org.aspectj.lang.annotation.Pointcut; import org.aspectj.lang.reflect.MethodSignature; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;/** * @author: hujunzheng * @create: 17/9/29 下午2:49 */ @Component @Aspect public class RedisLockAspect {private static final Log log = LogFactory.getLog(RedisLockAspect.class);@Autowiredprivate RedisCacheTemplate.RedisLockOperation redisLockOperation;@Pointcut("execution(* com.hjzgg..StargateDeployMessageConsumer.consumeStargateDeployMessage(..))" +"&& @annotation(me.ele.api.portal.service.redis.RedisLock)")private void lockPoint(){}@Around("lockPoint()")public Object arround(ProceedingJoinPoint pjp) throws Throwable{MethodSignature methodSignature = (MethodSignature) pjp.getSignature();Method method = methodSignature.getMethod();RedisLock lockInfo = method.getAnnotation(RedisLock.class);/*String lockKey = lockInfo.value();if (method.getParameters().length == 1 && pjp.getArgs()[0] instanceof DeployMessage) {DeployMessage deployMessage = (DeployMessage) pjp.getArgs()[0];lockKey += deployMessage.getEnv();System.out.println(lockKey);}*/boolean lock = false;Object obj = null;while(!lock){long timestamp = System.currentTimeMillis()+lockInfo.keepMills();lock = setNX(lockInfo.value(), timestamp);//得到鎖，已過期并且成功設置后舊的時間戳依然是過期的，可以認為獲取到了鎖(成功設置防止鎖競爭)long now = System.currentTimeMillis();if(lock || ((now > getLock(lockInfo.value())) && (now > getSet(lockInfo.value(), timestamp)))){log.info("得到redis分布式鎖...");obj = pjp.proceed();if(lockInfo.action().equals(RedisLock.LockFailAction.CONTINUE)){releaseLock(lockInfo.value());}}else{if(lockInfo.action().equals(RedisLock.LockFailAction.CONTINUE)){log.info("稍后重新請求redis分布式鎖...");Thread.currentThread().sleep(lockInfo.sleepMills());}else{log.info("放棄redis分布式鎖...");break;}}}return obj;}private boolean setNX(String key,Long value){return redisLockOperation.setNX(key, value);}private long getLock(String key){return redisLockOperation.get(key);}private Long getSet(String key,Long value){return redisLockOperation.getSet(key, value);}private void releaseLock(String key){redisLockOperation.delete(key);}@Pointcut(value = "execution(* me.ele..StargateBuildMessageConsumer.consumeStargateBuildMessage(me.ele.api.portal.service.mq.dto.BuildMessage)) && args(buildMessage)" +"&& @annotation(me.ele.api.portal.service.redis.RedisLock)", argNames = "buildMessage")private void buildMessageLockPoint(BuildMessage buildMessage){}@Around(value = "buildMessageLockPoint(buildMessage)", argNames = "pjp,buildMessage")public Object buildMessageAround(ProceedingJoinPoint pjp, BuildMessage buildMessage) throws Throwable {final String LOCK = buildMessage.getAppId() + buildMessage.getPushSequence();Lock lock = redisLockRegistry.obtain(LOCK);try {lock.lock();return pjp.proceed();} finally {try {lock.unlock();} catch (Exception e) {log.error("buildMessage={}, Lock {} unlock failed. {}", buildMessage, lock, e);}}}}

四、遇到的問題

　　

?

　　

　　開始是將鎖加到deploy的方法上的，但是一直aop一直沒有作用，換到consumeStargateDeployMessage方法上就可以了。考慮了一下是因為 @Transactional的原因。這里注意下。

?　　在一篇文章中找到了原因：SpringBoot CGLIB AOP解決Spring事務,對象調用自己方法事務失效.

　　只要脫離了Spring容器管理的所有對象,對于SpringAOP的注解都會失效,因為他們不是Spring容器的代理類,SpringAOP,就切入不了。也就是說是?@Transactional注解方法的代理對象并不是spring代理對象。

　　參考：?關于proxy模式下，@Transactional標簽在創建代理對象時的應用

五、使用spring-redis中的RedisLockRegistry

import java.util.concurrent.locks.Lock; import org.springframework.integration.redis.util.RedisLockRegistry;@Bean public RedisLockRegistry redisLockRegistry(@Value("${xxx.xxxx.registry}") String redisRegistryKey,RedisTemplate redisTemplate) {return new RedisLockRegistry(redisTemplate.getConnectionFactory(), redisRegistryKey, 200000); }Lock lock = redisLockRegistry.obtain(appId);lock.tryLock(180, TimeUnit.SECONDS); .... lock.unlock();　　

六、參考

　　其他工具類，請參考這里。

七、springboot LockRegistry

　　

? ? ??分布式鎖-RedisLockRegistry源碼分析[轉]

?

import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.locks.Lock; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.integration.redis.util.RedisLockRegistry; import redis.clients.jedis.JedisShardInfo;@Ignore public class RedisLockTest {private static final Logger LOGGER = LoggerFactory.getLogger(RedisLockTest.class);private static final String LOCK = "xxx.xxx";private RedisLockRegistry redisLockRegistry;@Beforepublic void setUp() {JedisShardInfo shardInfo = new JedisShardInfo("127.0.0.1");JedisConnectionFactory factory = new JedisConnectionFactory(shardInfo);redisLockRegistry = new RedisLockRegistry(factory, "test", 50L);}private class TaskA implements Runnable {@Overridepublic void run() {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}Lock lock = redisLockRegistry.obtain(LOCK);try {lock.lock();LOGGER.info("Lock {} is obtained", lock);Thread.sleep(10);lock.unlock();LOGGER.info("Lock {} is unlocked", lock);} catch (Exception ex) {LOGGER.error("Lock {} unlock failed", lock, ex);}}}private class TimeoutTask implements Runnable {@Overridepublic void run() {Lock lock = redisLockRegistry.obtain(LOCK);try {lock.lock();LOGGER.info("Lock {} is obtained", lock);Thread.sleep(5000);lock.unlock();LOGGER.info("Lock {} is unlocked", lock);} catch (Exception ex) {LOGGER.error("Lock {} unlock failed", lock, ex);}}}@Testpublic void test() throws InterruptedException, TimeoutException {ExecutorService service = Executors.newFixedThreadPool(2);service.execute(new TimeoutTask());service.execute(new TaskA());service.shutdown();if (!service.awaitTermination(1, TimeUnit.MINUTES)) {throw new TimeoutException();}} }

?

?

?

轉載于:https://www.cnblogs.com/hujunzheng/p/7612264.html

總結

以上是生活随笔為你收集整理的redis分布式锁小试的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。