跑流量

src/main/java/common/repository/TkioFlowRepository.java

@@ -24,4 +24,9 @@ public interface TkioFlowRepository  extends JpaRepository<TkioFlow, Long> {
 
     @Query(value = "select sum(flow) from tkio_flow where contract_code in ?3 and ds >= ?1 and ds <= ?2", nativeQuery = true)
     BigDecimal sumFlowByDsAndContractCodes(String startDate, String endDate, ArrayList<String> contractCode);
+
+    @Transactional
+    @Modifying
+    @Query(value = "delete from tkio_flow where email = ?1 and ds = ?2", nativeQuery = true)
+    void deleteByEmailDs(String email, String yesterday);
 }

src/main/java/security/interceptor/AuthorizationInterceptor.java

@@ -27,7 +27,6 @@ import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import java.io.BufferedWriter;
 import java.io.OutputStreamWriter;
-import java.util.Enumeration;
 import java.util.Map;
 import java.util.stream.Collectors;
 
@@ -61,8 +60,8 @@ public class AuthorizationInterceptor extends HandlerInterceptorAdapter {
         //放开登录
 
         String requestURL = request.getRequestURL().toString();
-        if(requestURL.contains("findSaleInfo")){
-            System.out.println("-----"+requestURL.contains("findSaleInfo"));
+        if (requestURL.contains("findSaleInfo")) {
+            System.out.println("-----" + requestURL.contains("findSaleInfo"));
             return true;
         }
 //        System.out.println("============="+requestURL);
@@ -72,15 +71,16 @@ public class AuthorizationInterceptor extends HandlerInterceptorAdapter {
                 || requestURL.indexOf("user/forget") != -1
                 || requestURL.indexOf("user/code") != -1
                 || requestURL.indexOf("txt2db") != -1
+                || requestURL.indexOf("/test/contract/flow") != -1
                 || requestURL.indexOf("contract/build") != -1
-                || requestURL.indexOf("accountmng/findSaleInfo") != -1){
+                || requestURL.indexOf("accountmng/findSaleInfo") != -1) {
             return true;
         }
 
         //从header中得到token
         String token = request.getHeader(httpHeaderName);
         //token = StringUtil.isEmpty(token) ? getCookieToken(request.getCookies(), "TOKEN") : token;
-        if(null != token){
+        if (null != token) {
             token = token.substring(httpHeaderPrefix.length());
             //验证token
             String key = manager.getKeyFromToken(token);
@@ -101,15 +101,15 @@ public class AuthorizationInterceptor extends HandlerInterceptorAdapter {
                         one.setAuthdataDic(authdata);
                     }
 
-                    request.getSession().setAttribute(Constant.CURRENT_ACCOUNT,one);
+                    request.getSession().setAttribute(Constant.CURRENT_ACCOUNT, one);
                     sessionAct = one;
                 }
 
-                if(!sessionAct.getRole().equals(RoleEnum.MANAGER.getKey())){
+                if (!sessionAct.getRole().equals(RoleEnum.MANAGER.getKey())) {
                     //权限控制
                     AuthKey authKey = handler.getClass().getAnnotation(AuthKey.class);
-                    if(authKey!=null && sessionAct.getAuthdataDic().get(authKey.value().getIdKey())==null){
-                        printJsonResponse(response,"权限不足!");
+                    if (authKey != null && sessionAct.getAuthdataDic().get(authKey.value().getIdKey()) == null) {
+                        printJsonResponse(response, "权限不足!");
                         return false;
                     }
 
@@ -121,31 +121,31 @@ public class AuthorizationInterceptor extends HandlerInterceptorAdapter {
 
         }
 
-        printJsonResponse(response,null);
+        printJsonResponse(response, null);
 
         return false;
     }
 
-    private void printJsonResponse(HttpServletResponse response,String message){
+    private void printJsonResponse(HttpServletResponse response, String message) {
         try {
             //如果验证token失败，并且方法注明了Authorization，返回401错误
             response.setStatus(HttpStatus.UNAUTHORIZED.value());
             BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(response.getOutputStream()));
             response.setContentType(MediaType.APPLICATION_JSON_VALUE);
 
-            if(StringUtils.isEmpty(message)){
+            if (StringUtils.isEmpty(message)) {
                 message = new ObjectMapper().writeValueAsString(ResultModel.ERROR(ResultStatus.USERNAME_LOGIN_EXPIRE));
             }
 
             writer.write(message);
             writer.close();
-        } catch (Exception e){
+        } catch (Exception e) {
             e.printStackTrace();
         }
     }
 
 
-    private String getCookieToken(Cookie[] cookies, String name){
+    private String getCookieToken(Cookie[] cookies, String name) {
 
         String token = null;
 

src/main/java/track/controller/TestFlowController.java

+package track.controller;
+
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RestController;
+import track.task.TrackingFlowTask;
+
+@RestController
+public class TestFlowController {
+
+
+    @Autowired
+    TrackingFlowTask trackingFlowTask;
+
+    @GetMapping("/test/contract/flow")
+    public void testFlow(String email) {
+        trackingFlowTask.reset(email);
+    }
+}

src/main/java/track/task/TrackingFlowTask.java

@@ -7,7 +7,9 @@ import common.repository.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.StopWatch;
 import tkio.model.Account;
 import tkio.repository.AccountRepository;
 import tkio.repository.AppRepository;
@@ -21,13 +23,18 @@ import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.text.ParseException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * @author liyin
  * @description
  * @date
  */
+@Service
 public class TrackingFlowTask {
 
     private final Logger logger = LoggerFactory.getLogger(TrackingFlowTask.class);
@@ -55,7 +62,16 @@ public class TrackingFlowTask {
      * 功能描述:定时同步流量(每7分钟执行一次)
      */
     public void syncFlow() {
-        List<CalculationFlow> calculationFlows = calculationFlowRepository.findByStatus(0);
+        syncFlow(null);
+    }
+
+    public void syncFlow(Long id) {
+        List<CalculationFlow> calculationFlows;
+        if (id != null) {
+            calculationFlows = Arrays.asList(calculationFlowRepository.findOne(id));
+        } else {
+            calculationFlows = calculationFlowRepository.findByStatus(0);
+        }
         for (CalculationFlow calculationFlow : calculationFlows) {
             List<TkioFlow> tkioFlowList = new ArrayList<>();
             calculationFlow.setStatus(1);
@@ -95,7 +111,7 @@ public class TrackingFlowTask {
                 for (int ii = between; ii > startInt; ii--) {
                     String yesterday = DateUtil.getBeforeDays(ii);//昨日
                     //BigInteger clickNum = accountFlowRestrictService.getTotalNum(yesterday, yesterday, appkeyStr, "account_track_flow_restrict", "click_sum");
-                    BigInteger clickNum = flowService.getFlowByAccount(yesterday,yesterday,appkeys);
+                    BigInteger clickNum = flowService.getFlowByAccount(yesterday, yesterday, appkeys);
                     if (clickNum != null && clickNum.longValue() > 0) {
                         List<Contract> contracts = contractRepository.findByPlatformAndEmail("tkio", email);
                         if (contracts.size() == 1) {//只有一个合同
@@ -214,8 +230,8 @@ public class TrackingFlowTask {
 
             //查询用户下所有appkey
             Account account = accountRepository.findByEmail(email);
-            if(account==null||account.getRootParent()==null){
-                logger.warn("【单日流量同步】用户不存在：{}",email);
+            if (account == null || account.getRootParent() == null) {
+                logger.warn("【单日流量同步】用户不存在：{}", email);
                 continue;
             }
             List<Account> accountList = accountRepository.findByRootParent(account.getRootParent());
@@ -226,12 +242,12 @@ public class TrackingFlowTask {
             List<String> appkeys = appRepository.findAppkeysNotDebug(idList);
             String appkeyStr = String.join("','", appkeys);
             appkeyStr = "'" + appkeyStr + "'";
-            if(CollectionUtils.isEmpty(appkeys)){
-                logger.warn("【单日流量同步】该用户没有appkey：{}",email);
+            if (CollectionUtils.isEmpty(appkeys)) {
+                logger.warn("【单日流量同步】该用户没有appkey：{}", email);
                 continue;
             }
             //BigInteger clickNum = accountFlowRestrictService.getTotalNum(yesterday, yesterday, appkeyStr, "account_track_flow_restrict", "click_sum");
-            BigInteger clickNum = flowService.getFlowByAccount(yesterday,yesterday,appkeys);
+            BigInteger clickNum = flowService.getFlowByAccount(yesterday, yesterday, appkeys);
             if (clickNum != null && clickNum.longValue() > 0) {
                 List<Contract> contracts = contractRepository.findByPlatformAndEmail("tkio", email);
                 if (contracts.size() == 1) {//只有一个合同
@@ -391,21 +407,63 @@ public class TrackingFlowTask {
     /**
      * 功能描述:临时跑一段时间内的流量任务
      */
-    public void reset(){
+    public void reset(String lastEmail) {
         List<String> emails = contractRepository.findDistinctEmailByPlatform("tkio");
+        List<String> dateInterval = DateUtil.getDateInterval("2021-05-25", "2021-07-15");
+        // dateInterval.add("2021-06-30");
+        // emails = Arrays.asList("1291269883@qq.com");
 
-        List<TkioFlow> tkioFlowList = new ArrayList<>();
-        //String yesterday = DateUtil.getBeforeDays(1);//昨日
-        List<String> dateInterval = DateUtil.getDateInterval("2021-05-25", "2021-06-06");
-        for (String yesterday : dateInterval) {
-            logger.info("昨日流量同步：" + yesterday);
+        if (lastEmail != null) {
+            List<String> unSyncEmails = new ArrayList<>();
+            boolean addflag = false;
             for (String email : emails) {
+                if (email != null && email.equals(lastEmail)) {
+                    addflag = true;
+                }
+
+                if (addflag) {
+                    unSyncEmails.add(email);
+                }
+            }
+
+            if (!unSyncEmails.isEmpty()) {
+                emails = unSyncEmails;
+            }
+        }
+
+
+        for (String email : emails) {
+            StopWatch stopWatch = new StopWatch();
+            stopWatch.start();
+            CompletableFuture[] futures = dateInterval.stream().map(dsone ->
+                    CompletableFuture.supplyAsync(() -> {
+                                reset(dsone, email);
+                                return Thread.currentThread().getName();
+                            }
+
+                    ).exceptionally((t) -> {
+                        logger.error("erro on " + email + "-" + dsone, t);
+                        return "-1";
+                    })).toArray(size -> new CompletableFuture[size]);
+            CompletableFuture.allOf(futures);
+            String names = Stream.of(futures).map(f -> f.join()).map(v -> ((String) v).split("-")[2]).collect(Collectors.joining(",-"));
+            stopWatch.stop();
+            logger.info(" {}s, emmial {} ,ForkJoinPool.commonPool-worker-{}", stopWatch.getTotalTimeSeconds(), email, names);
+        }
+        logger.info("reset complate ! ...");
+    }
+
+    public void reset(String yesterday, String email) {
+
+        //logger.info("ds {} eamil {} running...", yesterday, email);
+
+        tkioFlowRepository.deleteByEmailDs(email, yesterday);
 
         //查询用户下所有appkey
         Account account = accountRepository.findByEmail(email);
-                if(account==null||account.getRootParent()==null){
-                    logger.warn("【单日流量同步】用户不存在：{}",email);
-                    continue;
+        if (account == null || account.getRootParent() == null) {
+            logger.warn("【单日流量同步】用户不存在：{}", email);
+            return;
         }
         List<Account> accountList = accountRepository.findByRootParent(account.getRootParent());
         List<Long> idList = new ArrayList<>();
@@ -415,18 +473,18 @@ public class TrackingFlowTask {
         List<String> appkeys = appRepository.findAppkeysNotDebug(idList);
         String appkeyStr = String.join("','", appkeys);
         appkeyStr = "'" + appkeyStr + "'";
-                if(CollectionUtils.isEmpty(appkeys)){
-                    logger.warn("【单日流量同步】该用户没有appkey：{}",email);
-                    continue;
+        if (CollectionUtils.isEmpty(appkeys)) {
+            logger.warn("【单日流量同步】该用户没有appkey：{}", email);
+            return;
         }
         //BigInteger clickNum = accountFlowRestrictService.getTotalNum(yesterday, yesterday, appkeyStr, "account_track_flow_restrict", "click_sum");
-                BigInteger clickNum = flowService.getFlowByAccount(yesterday,yesterday,appkeys);
+        BigInteger clickNum = flowService.getFlowByAccount(yesterday, yesterday, appkeys);
         if (clickNum != null && clickNum.longValue() > 0) {
             List<Contract> contracts = contractRepository.findByPlatformAndEmail("tkio", email);
             if (contracts.size() == 1) {//只有一个合同
                 TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contracts.get(0));
                 if (tkioFlow != null) {
-                            tkioFlowList.add(tkioFlow);
+                    //tkioFlowList.add(tkioFlow);
                     tkioFlowRepository.save(tkioFlow);
                 }
             } else {//多个合同
@@ -452,21 +510,19 @@ public class TrackingFlowTask {
                     //第一优先级：合同开始日期，第二优先级，合同编号大小
                     Contract[] contractsArray = new Contract[correlationContract.size()];
                     contractsArray = correlationContract.toArray(contractsArray);
-
                     contractsArray = orderByContract(contractsArray);
-
                     for (int i = 0; i < contractsArray.length; i++) {
                         TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contractsArray[i]);
                         if (tkioFlow != null && i < contractsArray.length - 1 && tkioFlow.getCostFlow() != null && tkioFlow.getCostFlow() > 0L) {//处理成本流量，如果超出了，依次算在下一个合同上
                             clickNum = BigInteger.valueOf(tkioFlow.getCostFlow());
                             tkioFlow.setCostFlow(null);
                             if (tkioFlow.getFlow() > 0) {
-                                        tkioFlowList.add(tkioFlow);
+                                // tkioFlowList.add(tkioFlow);
                                 tkioFlowRepository.save(tkioFlow);
                             }
                         } else {
                             if (tkioFlow != null) {
-                                        tkioFlowList.add(tkioFlow);
+                                //tkioFlowList.add(tkioFlow);
                                 tkioFlowRepository.save(tkioFlow);
                             }
                             break;
@@ -494,7 +550,7 @@ public class TrackingFlowTask {
 
                                 TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contractsArray[j]);
                                 if (tkioFlow != null) {
-                                            tkioFlowList.add(tkioFlow);
+                                    //tkioFlowList.add(tkioFlow);
                                     tkioFlowRepository.save(tkioFlow);
                                 }
                                 break;
@@ -505,7 +561,7 @@ public class TrackingFlowTask {
                     } else {
                         TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, correlationContract.get(0));
                         if (tkioFlow != null) {
-                                    tkioFlowList.add(tkioFlow);
+                            //tkioFlowList.add(tkioFlow);
                             tkioFlowRepository.save(tkioFlow);
                         }
                     }
@@ -514,18 +570,14 @@ public class TrackingFlowTask {
             }
         }
 
-
-            }
     }
 
-
-    }
-    public static void main(String[] args) throws ParseException {
+   /* public static void main(String[] args) throws ParseException {
         String ago = "2020-10-01";
         int between = DateUtil.daysBetween(ago, DateUtil.getBeforeDays(1)) + 1;
         System.out.println(between);
         for (int i = between; i > 0; i--) {
             System.out.println(DateUtil.getBeforeDays(i));
         }
-    }
+    }*/
 }