<feat> gateway 为接口签名做准备

<feat> redisson 分布式锁改为看门狗式
<fix> DataStatisticUtil 对任务id做null判断
<fix> 任务调度部分改动

fanxing-commons/src/main/java/com/viontech/fanxing/commons/config/RedissonConfig.java

@@ -7,9 +7,7 @@ import org.redisson.spring.data.connection.RedissonConnectionFactory;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.core.io.Resource;
 
-import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 
@@ -32,6 +30,7 @@ public class RedissonConfig {
     public RedissonClient redisson(@Value("${vion.redisson.path}") String path) throws IOException {
         try (FileInputStream inputStream = new FileInputStream(path)) {
             Config config = Config.fromYAML(inputStream);
+            config.setLockWatchdogTimeout(12000);
             return Redisson.create(config);
         }
     }

fanxing-commons/src/main/java/com/viontech/fanxing/commons/service/RedisService.java

@@ -40,27 +40,18 @@ public class RedisService {
 
     public RLock getLockMust(String lockName) {
         RLock lock = redissonClient.getLock(lockName);
-        boolean isLock = false;
-        while (!isLock) {
-            try {
-                isLock = lock.tryLock(30, 25, TimeUnit.SECONDS);
-            } catch (Exception ignore) {
-            }
-        }
+        lock.lock();
         return lock;
     }
 
-    public RLock getLockMust(String lockName, Integer waitTime, Integer leaseTime, TimeUnit timeUnit) {
+    public RLock tryLock(String lockName) {
         RLock lock = redissonClient.getLock(lockName);
-        boolean isLock = false;
-        while (!isLock) {
-            try {
-                isLock = lock.tryLock(waitTime, leaseTime, timeUnit);
-                Thread.sleep(50L);
-            } catch (Exception ignore) {
-            }
+        boolean success = false;
+        try {
+            success = lock.tryLock(30, TimeUnit.SECONDS);
+        } catch (InterruptedException ignore) {
         }
-        return lock;
+        return success ? lock : null;
     }
 
     public RedissonClient getClient() {

fanxing-forward/src/main/java/com/viontech/fanxing/forward/util/DataStatisticUtil.java

@@ -62,6 +62,9 @@ public class DataStatisticUtil {
     }
 
     public void loadDataStatistic(Long taskId, String taskName, LocalDateTime dateTime, String redisKey) {
+        if (taskId == null) {
+            return;
+        }
         LocalDate localDate = dateTime.toLocalDate();
         int hour = dateTime.getHour();
         List<DataStatistic> query = jdbcTemplate.query("select * from d_data_statistic where date=? and task_id=?", new BeanPropertyRowMapper<>(DataStatistic.class), localDate, hour);

fanxing-gateway/src/main/java/com/viontech/fanxing/forward/AuthorizationFilter.java

@@ -11,6 +11,7 @@ import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.springframework.cloud.gateway.filter.GatewayFilterChain;
 import org.springframework.cloud.gateway.filter.GlobalFilter;
+import org.springframework.core.Ordered;
 import org.springframework.http.*;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
@@ -37,7 +38,7 @@ import java.util.concurrent.Future;
 
 @Slf4j
 @Component
-public class AuthorizationFilter implements GlobalFilter {
+public class AuthorizationFilter implements GlobalFilter, Ordered {
 
     ExecutorService executorService = Executors.newCachedThreadPool();
     @Resource
@@ -161,4 +162,9 @@ public class AuthorizationFilter implements GlobalFilter {
         }
         return null;
     }
+
+    @Override
+    public int getOrder() {
+        return HIGHEST_PRECEDENCE + 101;
+    }
 }

fanxing-gateway/src/main/java/com/viontech/fanxing/forward/CacheBodyGatewayFilter.java

+package com.viontech.fanxing.forward;
+
+import org.springframework.cloud.gateway.filter.GatewayFilterChain;
+import org.springframework.cloud.gateway.filter.GlobalFilter;
+import org.springframework.core.Ordered;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * .
+ *
+ * @author 谢明辉
+ * @date 2022/2/17
+ */
+
+//@Component
+public class CacheBodyGatewayFilter implements Ordered, GlobalFilter {
+
+    public static final String CACHE_REQUEST_BODY_OBJECT_KEY = "cachedRequestBodyObject";
+
+    @Override
+    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
+        if (exchange.getRequest().getHeaders().getContentType() == null) {
+            return chain.filter(exchange);
+        } else {
+            return DataBufferUtils.join(exchange.getRequest().getBody()).flatMap(dataBuffer -> {
+                DataBufferUtils.retain(dataBuffer);
+                Flux<DataBuffer> cachedFlux = Flux.defer(() -> Flux.just(dataBuffer.slice(0, dataBuffer.readableByteCount())));
+                ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator(exchange.getRequest()) {
+                    @Override
+                    public Flux<DataBuffer> getBody() {
+                        return cachedFlux;
+                    }
+
+                };
+                exchange.getAttributes().put(CACHE_REQUEST_BODY_OBJECT_KEY, cachedFlux);
+
+                return chain.filter(exchange.mutate().request(mutatedRequest).build());
+            });
+        }
+    }
+
+    @Override
+    public int getOrder() {
+        return Ordered.HIGHEST_PRECEDENCE;
+    }
+}

fanxing-query/src/main/java/com/viontech/fanxing/query/runner/ExportDataJob.java

@@ -43,7 +43,6 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.TimeUnit;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipOutputStream;
 
@@ -85,7 +84,10 @@ public class ExportDataJob {
     @Scheduled(fixedDelay = 10000L)
     public void run() {
         log.info("执行数据导出定时任务开始");
-        RLock lock = redisService.getLockMust("exportDataJob", 60, 59, TimeUnit.MINUTES);
+        RLock lock = redisService.tryLock("exportDataJob");
+        if (lock == null) {
+            return;
+        }
         try {
             List<ExportData> exportData = getExportDataNeedFinished();
             for (ExportData item : exportData) {
@@ -213,8 +215,7 @@ public class ExportDataJob {
     }
 
 
-    private <K> ImmutableTriple<Boolean, String, Integer> export(ExportData item, Function<Integer, PageInfo<K>> getJsonData, Function<K, ExportBaseModel> build, boolean withPic, boolean withVideo, Integer nextPage,
-                                                                 Function<K, String> getPic, Function<K, String> getVideo) {
+    private <K> ImmutableTriple<Boolean, String, Integer> export(ExportData item, Function<Integer, PageInfo<K>> getJsonData, Function<K, ExportBaseModel> build, boolean withPic, boolean withVideo, Integer nextPage, Function<K, String> getPic, Function<K, String> getVideo) {
         if (needInterrupt(item.getId())) {
             log.info("任务已被删除,打断任务");
             return ImmutableTriple.of(false, null, null);
@@ -301,10 +302,7 @@ public class ExportDataJob {
         int nextPage = 1;
         List<String> pathList = new ArrayList<>();
         while (nextPage != 0) {
-            ImmutableTriple<Boolean, String, Integer> export = export(item,
-                    (page) -> trafficService.getJsonData(example, page, pageSize),
-                    x -> buildTrafficData(type, x), withPic, withVideo, nextPage,
-                    TrafficVo::getPics, TrafficVo::getVideoName);
+            ImmutableTriple<Boolean, String, Integer> export = export(item, (page) -> trafficService.getJsonData(example, page, pageSize), x -> buildTrafficData(type, x), withPic, withVideo, nextPage, TrafficVo::getPics, TrafficVo::getVideoName);
             if (export.left) {
                 String path = export.middle;
                 pathList.add(path);
@@ -334,10 +332,7 @@ public class ExportDataJob {
         int nextPage = 1;
         List<String> pathList = new ArrayList<>();
         while (nextPage != 0) {
-            ImmutableTriple<Boolean, String, Integer> export = export(item,
-                    (page) -> behaviorService.getJsonData(example, page, pageSize),
-                    this::buildBehaviorModel, withPic, withVideo, nextPage,
-                    BehaviorVo::getPics, BehaviorVo::getVideo);
+            ImmutableTriple<Boolean, String, Integer> export = export(item, (page) -> behaviorService.getJsonData(example, page, pageSize), this::buildBehaviorModel, withPic, withVideo, nextPage, BehaviorVo::getPics, BehaviorVo::getVideo);
             if (export.left) {
                 String path = export.middle;
                 pathList.add(path);

fanxing-task/src/main/java/com/viontech/fanxing/task/runner/TaskRunner.java

@@ -41,9 +41,12 @@ public class TaskRunner {
     @Resource
     private TaskService taskService;
 
-    @Scheduled(fixedDelay = 5000)
+    @Scheduled(fixedDelay = 2000)
     public void run() {
-        RLock jobLock = redisService.getLockMust("lock:taskRunner");
+        RLock jobLock = redisService.tryLock("lock:taskRunner");
+        if (jobLock == null) {
+            return;
+        }
         try {
             terminatedTask();
             executedTask();
@@ -127,8 +130,8 @@ public class TaskRunner {
         try {
             RScoredSortedSet<String> set = redisService.getToBeTerminatedTaskUnidSet();
             RScoredSortedSet<String> toBeExecutedTaskUnidSet = redisService.getToBeExecutedTaskUnidSet();
-            // 提前 1 秒预读
-            Collection<String> entryCollection = set.valueRange(0, true, System.currentTimeMillis() + 1000, true);
+            // 提前 2 秒预读
+            Collection<String> entryCollection = set.valueRange(0, true, System.currentTimeMillis() + 2000, true);
 
             for (String taskUnid : entryCollection) {
                 TaskData taskData = taskDataService.getRepository().getTaskDataByUnid(taskUnid);

fanxing-task/src/main/java/com/viontech/fanxing/task/service/VAServerService.java

@@ -114,6 +114,11 @@ public class VAServerService {
         // 如果vaServerId不为空,需要终止任务
         if (vaServerId != null) {
             VaServerInfo vaServerInfo = vaServerRedisRepository.getVAServerInfoById(vaServerId);
+            // 如果找不到对应的 vaServer , 相当于已经停止了任务
+            if (vaServerInfo == null) {
+                return true;
+            }
+            // 如果 vaServer 为离线, 那么无法停止
             if (0 == vaServerInfo.getStatus()) {
                 return false;
             }