如何操作Redis和zookeeper實(shí)現(xiàn)分布式鎖

在分布式場景下，有很多種情況都需要實(shí)現(xiàn)最終一致性。在設(shè)計遠(yuǎn)程上下文的領(lǐng)域事件的時候，為了保證最終一致性，在通過領(lǐng)域事件進(jìn)行通訊的方式中，可以共享存儲（領(lǐng)域模型和消息的持久化數(shù)據(jù)源），或者做全局XA事務(wù)（兩階段提交，數(shù)據(jù)源可分開），也可以借助消息中間件（消費(fèi)者處理需要能冪等）。通過Observer模式來發(fā)布領(lǐng)域事件可以提供很好的高并發(fā)性能，并且事件存儲也能追溯更小粒度的事件數(shù)據(jù)，使各個應(yīng)用系統(tǒng)擁有更好的自治性。

1.分布式鎖

分布式鎖一般用在分布式系統(tǒng)或者多個應(yīng)用中，用來控制同一任務(wù)是否執(zhí)行或者任務(wù)的執(zhí)行順序。在項(xiàng)目中，部署了多個tomcat應(yīng)用，在執(zhí)行定時任務(wù)時就會遇到同一任務(wù)可能執(zhí)行多次的情況，我們可以借助分布式鎖，保證在同一時間只有一個tomcat應(yīng)用執(zhí)行了定時任務(wù)。

2.分布式鎖的實(shí)現(xiàn)方式

• 使用redis的setnx()和expire()
• 使用redis的getset()
• 使用zookeeper的創(chuàng)建節(jié)點(diǎn)node
• 使用zookeeper的創(chuàng)建臨時序列節(jié)點(diǎn)
  

3.使用redis的setnx()和expire()來實(shí)現(xiàn)分布式鎖

setnx(key,value) 如果key不存在，設(shè)置為當(dāng)前key的值為value;如果key存在，直接返回。
expire()來設(shè)置超時時間

定義注解類：

@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface Lockable{
  // redis緩存key
  String key();
  // redis緩存key中的數(shù)據(jù)
  String value() default "";
  // 過期時間(秒)，默認(rèn)為一分鐘
  long expire() default 60;
}

定時任務(wù)增加注解@Lockable：

 @Lockable(key = "DistributedLock:dealExpireRecords")
 public void dealExpireRecords() {
 }

定義一個aop切面LockAspect，使用@Around處理所有注解為@Lockable的方法，通過連接點(diǎn)確認(rèn)此注解是用在方法上，通過方法獲取注解信息，使用setIfAbsent來判斷是否獲取分布式鎖，如果沒有獲取分布式鎖，直接返回；如果獲取到分布式鎖，通過expire設(shè)置過期時間，并調(diào)用指定方法。

@Component
@Slf4j
@Aspect
public class LockAspect {

  @Autowired
  private RedisTemplate redisTemplate;

  @Around("@annotation(com.records.aop.Lockable)")
  public Object distributeLock(ProceedingJoinPoint pjp) {
    Object resultObject = null;

    //確認(rèn)此注解是用在方法上
    Signature signature = pjp.getSignature();
    if (!(signature instanceof MethodSignature)) {
      log.error("Lockable is method annotation!");
      return resultObject;
    }

    MethodSignature methodSignature = (MethodSignature) signature;
    Method targetMethod = methodSignature.getMethod();

    //獲取注解信息
    Lockable lockable = targetMethod.getAnnotation(Lockable.class);
    String key = lockable.key();
    String value = lockable.value();
    long expire = lockable.expire();

    // 分布式鎖，如果沒有此key，設(shè)置此值并返回true；如果有此key，則返回false
    boolean result = redisTemplate.boundValueOps(key).setIfAbsent(value);
    if (!result) {
      //其他程序已經(jīng)獲取分布式鎖
      return resultObject;
    }

    //設(shè)置過期時間，默認(rèn)一分鐘
    redisTemplate.boundValueOps(key).expire(expire, TimeUnit.SECONDS);

    try {
      resultObject = pjp.proceed(); //調(diào)用對應(yīng)方法執(zhí)行
    } catch (Throwable throwable) {
      throwable.printStackTrace();
    }
    return resultObject;
  }
}

4.使用redis的getset()來實(shí)現(xiàn)分布式鎖

此方法使redisTemplate.boundValueOps(key).getAndSet(value)的方法，如果返回空，表示獲取了分布式鎖；如果返回不為空，表示分布式鎖已經(jīng)被其他程序占用

5.使用zookeeper的創(chuàng)建節(jié)點(diǎn)node

使用zookeeper創(chuàng)建節(jié)點(diǎn)node，如果創(chuàng)建節(jié)點(diǎn)成功，表示獲取了此分布式鎖；如果創(chuàng)建節(jié)點(diǎn)失敗，表示此分布式鎖已經(jīng)被其他程序占用(多個程序同時創(chuàng)建一個節(jié)點(diǎn)node，只有一個能夠創(chuàng)建成功)

6.使用zookeeper的創(chuàng)建臨時序列節(jié)點(diǎn)

使用zookeeper創(chuàng)建臨時序列節(jié)點(diǎn)來實(shí)現(xiàn)分布式鎖，適用于順序執(zhí)行的程序，大體思路就是創(chuàng)建臨時序列節(jié)點(diǎn)，找出最小的序列節(jié)點(diǎn)，獲取分布式鎖，程序執(zhí)行完成之后此序列節(jié)點(diǎn)消失，通過watch來監(jiān)控節(jié)點(diǎn)的變化，從剩下的節(jié)點(diǎn)的找到最小的序列節(jié)點(diǎn)，獲取分布式鎖，執(zhí)行相應(yīng)處理，依次類推......

感謝閱讀，希望能幫助到大家，謝謝大家對本站的支持！