[chg]：定时任务增加分布式锁改造，支持集群部署

src/main/java/com/viontech/keliu/cron/TopicCleanupJob.java

@@ -4,7 +4,9 @@ import cn.hutool.core.date.DateTime;
 import cn.hutool.core.date.DateUtil;
 import com.viontech.keliu.service.KafkaTopicService;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
@@ -12,6 +14,7 @@ import javax.annotation.Resource;
 import java.time.LocalDate;
 import java.time.ZoneId;
 import java.util.Date;
+import java.util.concurrent.TimeUnit;
 
 /**
  * kafka topic 清理任务
@@ -25,9 +28,19 @@ public class TopicCleanupJob {
     private Integer preDay;
     @Resource
     private KafkaTopicService kafkaTopicService;
+    @Autowired
+    private RedisTemplate redisTemplate;
+    // TopicCleanupJob的分布式锁
+    private static final String TOPICCLEANUPJOB_LOCK_KEY = "lock:TopicCleanupJob:";
 
     @Scheduled(cron = "${vion.topicCleanup.cron:-}")
     public void doTopicCleanup() {
+        String lockKey = TOPICCLEANUPJOB_LOCK_KEY + DateUtil.format(new Date(), "yyyyMMdd");
+        Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", 2, TimeUnit.DAYS);
+        log.info("doTopicCleanup.lockKey {}={}", lockKey, locked);
+        if (locked != null && !locked) {
+            return;
+        }
         DateTime preDate = DateUtil.offsetDay(new Date(), preDay * (-1));
         DateTime countDate = DateUtil.beginOfDay(preDate);
         LocalDate maxCountDate = countDate.toInstant().atZone(ZoneId.of("Asia/Shanghai")).toLocalDate();