参见 https://www.cnblogs.com/austinspark-jessylu/p/8043726.html

场景

任务中心：定时调度执行N多任务。

需要有个单机分发机器来生成任务列表并分发任务，怎么保证单机执行？

采用缓存

基本思路：利用缓存中存储一个特定的key来是否抢到了锁。利用缓存的版本机制，同一个版本只能存储一次数据，否则返回失败。

自己的场景实践—存在问题（见代码中描述）

1、每台机器先获取锁

  • 先查询是否tair value是否为空
  • 为空，直接put进去，抢到了锁
  • 不为空，则判断是否锁超时（因为是定时调度，保证本次调度内单机执行，下次调度重新抢锁）
    • 如果超时，则重新put新值进去，相当于抢锁
    • 否则，未抢到锁，相当于其他机器抢到了锁去执行

2、抢到锁 执行业务逻辑

3、记得释放锁

  1. @Override
  2. public void createInstanceByScheduler() throws Exception {
  4.     /** 0 获取tair锁 */
  5.     boolean tairFlag = middleWareService.updateCurrentHost();
  6.     if (!tairFlag) {
  7.         // 锁获取失败直接退出
  8.         throw new Exception("tairLock has been uesd");
  9.     }
  11.     try {
  12.         /** 1 拉取需要执行的任务 */
  13.         List<CommonTaskMainInfo> taskList = queryTaskToExecute();
  14.         if (CollectionUtils.isEmpty(taskList)) { return; }
  16.         /** 2 放入PriorityQueue中 */
  17.         PriorityBlockingQueue taskQueue = new PriorityBlockingQueue(taskList.size());
  18.         taskList.forEach(taskMainInfo -> taskQueue.put(taskMainInfo));
  20.         /** 3 异步线程消费PriorityQueue,生成instance,放入msgBroker */
  21.         consumeQueue(taskQueue, taskList);
  23.         /** 4 更新task调度信息 */
  24.         updateExecuteInfoAndSave(taskList);
  25.     } catch (Exception e) {
  26.         throw e;
  27.     } finally {
  28.         /** 5 归还tair锁 */
  29.         middleWareService.clearCurrentHost();
  30.     }
  31.     }
  34. @Override
  35.     public ScheduleHost queryCurrentHost() {
  36.         Object result = tairCacheClient.getObject(TAIR_KEY);
  38.         if (result == null) {
  39.             return null;
  40.         } else {
  41.             return (ScheduleHost) result;
  42.         }
  43.     }
  45. @Override
  46. public boolean updateCurrentHost() {
  47.     ScheduleHost currentHost = queryCurrentHost();
  49.     // 以下的写法存在问题，多机器并发执行，需要双重判空
  50.     if (currentHost != null && !currentHost.isEmpty()) {
  51.         long scheduleTime = currentHost.getScheduleTime().getTime();
  52.         long runTime = new Date().getTime() - scheduleTime;
  54.         // 如果tair锁被占用超过3分钟没有归还，将被清空
  55.         if (runTime < EXPIRED_TIME) {
  56.             LogUtil.info(LOGGER, "[helios] updateCurrentHost error, currentHost={0}", currentHost);
  57.             return false;
  58.         }
  59.     }
  61.     ScheduleHost host = new ScheduleHost(new Date(), EnvUtil.getHostName());
  62.     tairCacheClient.putObject(TAIR_KEY, host);
  63.     return true;
  64. }
  66. @Override
  67. public void clearCurrentHost() {
  68.     tairCacheClient.putObject(TAIR_KEY, new ScheduleHost());
  69. }

业界标准实现

image.png

2和0分别为版本号和过期时间，0代表不过期

采用DB

原理：利用数据库的特性 —— 唯一键，同一个键 只有一条记录能插入成功，来抢锁

  1. // 锁类型/场景
  2. private int    type;
  3. // 唯一键标识
  4. private String lockInfo;
  6. DataBaseLockDO lock = new DataBaseLockDO();
  7. lock.setType(2);
  8. lock.setLockInfo("feedbackReport#"
  9.                  + reportDO.getPlatform()
  10.                  + "#"
  11.                  + reportDO.getBizIndex()
  12.                  + "#"
  13.                  + reportDO.getSubTypeIndex()
  14.                  + "#"
  15.                  + DateUtil.formatDate(new Date(currentTime),
  16.                      DateUtil.DEFAULT_HOUR_STYLE_NOSPACE));
  17. // 能插入成功才继续
  18. if (!this.dataBaseLockDAO.insert(lock)) {
  19.     return;
  20. }