Merge pull request #272 from facming/feature/xxl-job

Feature/xxl job
Dapp-Learning-DAO · Jul 28, 2024 · 5ee0807 · 5ee0807
2 parents d4f2e81 + 38f47a0
commit 5ee0807
Show file tree

Hide file tree

Showing 14 changed files with 346 additions and 24 deletions.
diff --git a/build.gradle b/build.gradle
@@ -64,6 +64,7 @@ dependencies{
 	implementation 'io.springfox:springfox-boot-starter:3.0.0'
     implementation 'com.qcloud:cos_api:5.6.155'
     implementation 'org.springframework.social:spring-social-twitter:1.1.2.RELEASE'
+	implementation("com.xuxueli:xxl-job-core:2.4.1")
 
 	//graphql
 //	implementation(platform("com.netflix.graphql.dgs:graphql-dgs-platform-dependencies:latest.release"))

diff --git a/src/main/java/com/dl/officialsite/bounty/BountyService.java b/src/main/java/com/dl/officialsite/bounty/BountyService.java
@@ -23,6 +23,9 @@
 import java.util.LinkedList;
 import java.util.List;
 import javax.persistence.criteria.Predicate;
+
+import com.xxl.job.core.context.XxlJobHelper;
+import com.xxl.job.core.handler.annotation.XxlJob;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeanUtils;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -67,10 +70,8 @@ public BountyService(BountyRepository bountyRepository,
 
 
 
-    @Scheduled(cron = "${jobs.bounty.corn:0 0 * * * *}")
-    @ConditionalOnProperty(name = "scheduler.enabled", havingValue = "true", matchIfMissing = true)
     public void updateBountyData() {
-        log.info("updateBountyData_schedule task begin --------------------- ");
+        XxlJobHelper.log("updateBountyData_schedule task begin --------------------- ");
         //update status
         long currentSeconds = System.currentTimeMillis() / 1000;
         List<Bounty> bountyList = bountyRepository.findAll(

diff --git a/src/main/java/com/dl/officialsite/bounty/job/BountyXxlJob.java b/src/main/java/com/dl/officialsite/bounty/job/BountyXxlJob.java
@@ -0,0 +1,29 @@
+package com.dl.officialsite.bounty.job;
+
+import com.dl.officialsite.bounty.BountyService;
+import com.xxl.job.core.context.XxlJobHelper;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+/**
+ * @Description
+ * @Author xiaoming
+ * @Date 2024/7/6 7:45 PM
+ **/
+@Component
+public class BountyXxlJob {
+
+    @Autowired
+    private BountyService bountyService;
+
+    /**
+     * 更新bounty数据
+     */
+    @XxlJob("updateBountyDataJobHandler")
+    public void updateBountyStatus() {
+        XxlJobHelper.log("updateBountyData start");
+        bountyService.updateBountyData();
+        XxlJobHelper.log("updateBountyData end");
+    }
+}
diff --git a/src/main/java/com/dl/officialsite/config/XxlJobConfig.java b/src/main/java/com/dl/officialsite/config/XxlJobConfig.java
@@ -0,0 +1,37 @@
+package com.dl.officialsite.config;
+
+import com.xxl.job.core.executor.XxlJobExecutor;
+import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @Description
+ * @Author xiaoming
+ * @Date 2024/7/5 10:07 PM
+ **/
+@Configuration
+@Slf4j
+@EnableConfigurationProperties({XxlJobProperties.class})
+public class XxlJobConfig {
+
+    @Bean
+    public XxlJobExecutor xxlJobExecutor(XxlJobProperties properties) {
+        log.info("[xxlJobExecutor][初始化 XXL-Job 执行器的配置]");
+        XxlJobProperties.AdminProperties admin = properties.getAdmin();
+        XxlJobProperties.ExecutorProperties executor = properties.getExecutor();
+
+        // 初始化执行器
+        XxlJobExecutor xxlJobExecutor = new XxlJobSpringExecutor();
+        xxlJobExecutor.setIp(executor.getIp());
+        xxlJobExecutor.setPort(executor.getPort());
+        xxlJobExecutor.setAppname(executor.getAppName());
+        xxlJobExecutor.setLogPath(executor.getLogPath());
+        xxlJobExecutor.setLogRetentionDays(executor.getLogRetentionDays());
+        xxlJobExecutor.setAdminAddresses(admin.getAddresses());
+        xxlJobExecutor.setAccessToken(properties.getAccessToken());
+        return xxlJobExecutor;
+    }
+}
diff --git a/src/main/java/com/dl/officialsite/config/XxlJobProperties.java b/src/main/java/com/dl/officialsite/config/XxlJobProperties.java
@@ -0,0 +1,95 @@
+package com.dl.officialsite.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+import javax.validation.Valid;
+import javax.validation.constraints.NotEmpty;
+import javax.validation.constraints.NotNull;
+
+/**
+ * XXL-Job 配置类
+ */
+@ConfigurationProperties("xxl.job")
+@Validated
+@Data
+public class XxlJobProperties {
+
+    /**
+     * 访问令牌
+     */
+    private String accessToken;
+    /**
+     * 控制器配置
+     */
+    @NotNull(message = "控制器配置不能为空")
+    private AdminProperties admin;
+    /**
+     * 执行器配置
+     */
+    @NotNull(message = "执行器配置不能为空")
+    private ExecutorProperties executor;
+
+    /**
+     * XXL-Job 调度器配置类
+     */
+    @Data
+    @Valid
+    public static class AdminProperties {
+
+        /**
+         * 调度器地址
+         */
+        @NotEmpty(message = "调度器地址不能为空")
+        private String addresses;
+
+    }
+
+    /**
+     * XXL-Job 执行器配置类
+     */
+    @Data
+    @Valid
+    public static class ExecutorProperties {
+
+        /**
+         * 默认端口
+         *
+         * 这里使用 -1 表示随机
+         */
+        private static final Integer PORT_DEFAULT = -1;
+
+        /**
+         * 默认日志保留天数
+         *
+         * 如果想永久保留，则设置为 -1
+         */
+        private static final Integer LOG_RETENTION_DAYS_DEFAULT = 30;
+
+        /**
+         * 应用名
+         */
+        @NotEmpty(message = "应用名不能为空")
+        private String appName;
+        /**
+         * 执行器的 IP
+         */
+        private String ip;
+        /**
+         * 执行器的 Port
+         */
+        private Integer port = PORT_DEFAULT;
+        /**
+         * 日志地址
+         */
+        @NotEmpty(message = "日志地址不能为空")
+        private String logPath;
+        /**
+         * 日志保留天数
+         */
+        private Integer logRetentionDays = LOG_RETENTION_DAYS_DEFAULT;
+
+    }
+
+}
diff --git a/src/main/java/com/dl/officialsite/defi/Schedule.java b/src/main/java/com/dl/officialsite/defi/Schedule.java
@@ -9,6 +9,8 @@
 import com.dl.officialsite.team.vo.TeamsWithMembers;
 import java.math.BigDecimal;
 import java.util.List;
+
+import com.xxl.job.core.context.XxlJobHelper;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Scheduled;
@@ -38,7 +40,6 @@ public class Schedule {
     /**
      * 监控价格一天发送一次
      */
-    @Scheduled(cron = "0 0 12 * * ?")
     public void monitorPrice() throws Exception {
         log.info("monitorPrice start");
         //查找team0 memeber获取地址
@@ -60,10 +61,9 @@ public void monitorPrice() throws Exception {
     /**
      * 监控健康系数，如果小于1.2，立即发送邮件
      */
-    @Scheduled(cron = "0 0/30 * * * ? ")
     public void monitorHealth() {
         try {
-            log.info("monitorHealth start");
+            XxlJobHelper.log("monitorHealth start");
             //查找team0 memeber获取地址
             TeamQueryVo teamQueryVo = new TeamQueryVo();
             teamQueryVo.setTeamName("Dapp-Learning DAO core founders");
@@ -85,7 +85,7 @@ public void monitorHealth() {
                 }
             }
         } catch (Exception e) {
-            log.error("monitorHealth error", e);
+            XxlJobHelper.log("monitorHealth error", e);
         }
     }
 

diff --git a/src/main/java/com/dl/officialsite/defi/job/DefiXxlJob.java b/src/main/java/com/dl/officialsite/defi/job/DefiXxlJob.java
@@ -0,0 +1,84 @@
+package com.dl.officialsite.defi.job;
+
+import com.dl.officialsite.defi.Schedule;
+import com.dl.officialsite.defi.TokenInfoList;
+import com.dl.officialsite.defi.service.AaveService;
+import com.dl.officialsite.defi.service.AaveTokenAPYService;
+import com.dl.officialsite.defi.service.WhaleService;
+import com.dl.officialsite.defi.vo.HealthInfoVo;
+import com.dl.officialsite.mail.EmailService;
+import com.dl.officialsite.member.Member;
+import com.dl.officialsite.team.TeamService;
+import com.dl.officialsite.team.vo.TeamQueryVo;
+import com.dl.officialsite.team.vo.TeamsWithMembers;
+import com.xxl.job.core.context.XxlJobHelper;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.math.BigDecimal;
+import java.util.List;
+
+/**
+ * @Description
+ * @Author xiaoming
+ * @Date 2024/7/6 7:02 PM
+ **/
+@Component
+@Slf4j
+public class DefiXxlJob {
+
+    @Autowired
+    private Schedule schedule;
+
+    @Autowired
+    private AaveTokenAPYService aaveTokenAPYService;
+
+    @Autowired
+    private WhaleService whaleService;
+
+
+    /**
+     * 监控价格一天发送一次
+     */
+    @XxlJob("monitorPriceJobHandler")
+    public void monitorPrice() throws Exception {
+        XxlJobHelper.log("monitorPrice start");
+        schedule.monitorPrice();
+        XxlJobHelper.log("monitorPrice end");
+    }
+
+    /**
+     * 监控健康系数，如果小于1.2，立即发送邮件
+     */
+    @XxlJob("monitorHealthJobHandler")
+    public void monitorHealth() {
+        XxlJobHelper.log("monitorHealth start");
+        schedule.monitorHealth();
+        XxlJobHelper.log("monitorHealth end");
+    }
+
+
+    /**
+     * 定期更新 Token 的 APY（年化收益率）信息
+     */
+    @XxlJob("updateTokenAPYInfoJobHandler")
+    public void updateTokenAPYInfo()  {
+        XxlJobHelper.log("update token info task begin --------------------- ");
+        aaveTokenAPYService.updateTokenAPYInfo();
+        XxlJobHelper.log("update token info task end --------------------- ");
+    }
+
+    /**
+     * 更新Whale和WhaleTxRow数据
+     */
+    @XxlJob("aaveListenerJobHandler")
+    public void aaveListener() {
+        XxlJobHelper.log("aaveListenerJobHandler start");
+        whaleService.aaveListener();
+        XxlJobHelper.log("aaveListenerJobHandler end");
+    }
+
+}
diff --git a/src/main/java/com/dl/officialsite/defi/service/AaveTokenAPYService.java b/src/main/java/com/dl/officialsite/defi/service/AaveTokenAPYService.java
@@ -21,6 +21,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
+
+import com.xxl.job.core.context.XxlJobHelper;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.scheduling.annotation.Scheduled;
@@ -131,12 +133,12 @@ public List<ChainInfo> queryChainList() {
         return new ArrayList<>(Web3jAutoConfiguration.web3jMap.keySet());
     }
 
-    @Scheduled(cron =  "${jobs.defi.corn: 0 30 * * * * ?}")
-    @ConditionalOnProperty(name = "scheduler.enabled", havingValue = "true", matchIfMissing = true)
     public void updateTokenAPYInfo()  {
         DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
         log.info("update token info task begin --------------------- ");
+        XxlJobHelper.log("update token info task begin --------------------- ");
         log.info("now date {}", LocalDateTime.now().format(formatter));
+        XxlJobHelper.log("now date {}", LocalDateTime.now().format(formatter));
         List<TokenAPYInfo> tokenAPYInfoList = queryTokenApyOnChain();
         tokenAPYInfoRepository.deleteAll();
         tokenAPYInfoRepository.saveAll(tokenAPYInfoList);

diff --git a/src/main/java/com/dl/officialsite/defi/service/WhaleService.java b/src/main/java/com/dl/officialsite/defi/service/WhaleService.java
@@ -27,6 +27,8 @@
 import javax.persistence.criteria.CriteriaQuery;
 import javax.persistence.criteria.Predicate;
 import javax.persistence.criteria.Root;
+
+import com.xxl.job.core.context.XxlJobHelper;
 import lombok.extern.slf4j.Slf4j;
 import okhttp3.HttpUrl;
 import okhttp3.MediaType;
@@ -86,8 +88,6 @@ public WhaleService(WhaleRepository whaleRepository, WhaleTxRowRepository whaleT
         this.whaleChainTokenRepository = whaleChainTokenRepository;
     }
 
-    @Scheduled(cron =  "${jobs.defi.corn: 0 30 * * * * ?}")
-    @ConditionalOnProperty(name = "scheduler.enabled", havingValue = "true", matchIfMissing = true)
     public void aaveListener() {
         List<Whale> whaleList = new ArrayList<>();
         List<WhaleTxRow> insertWhaleTxRowList = new ArrayList<>();
@@ -102,6 +102,7 @@ public void aaveListener() {
         String jsonStr = requestAaveGraph(100, 0);
         JSONObject jsonObject = JSONUtil.parseObj(jsonStr);
         if (jsonObject.toString().contains("error")) {
+            XxlJobHelper.log("解析aave的graph失败");
             throw new RuntimeException("解析aave的graph失败");
         }
         JSONObject data = jsonObject.getJSONObject("data");
@@ -134,8 +135,11 @@ public void aaveListener() {
 
     public void insertWhaleAndTx(List<Whale> whaleList, List<WhaleTxRow> whaleTxRowList) {
         log.info("开始插入数据");
+        XxlJobHelper.log("开始插入数据");
         log.info("whaleList.size() = {}-------------------", whaleList.size());
+        XxlJobHelper.log("whaleList.size() = {}-------------------", whaleList.size());
         log.info("whaleTxRowList.size() = {}-------------------", whaleTxRowList.size());
+        XxlJobHelper.log("whaleTxRowList.size() = {}-------------------", whaleTxRowList.size());
         if (!whaleList.isEmpty()) {
             batchRepository.batchInsert(whaleList);
         }
@@ -156,6 +160,7 @@ public void insertWhaleAndTx(List<Whale> whaleList, List<WhaleTxRow> whaleTxRowL
             }
         }
         log.info("插入数据结束");
+        XxlJobHelper.log("插入数据结束");
     }
 
     private Long getOneYearBefore() {