限流器

job-bytedance/pom.xml

@@ -183,6 +183,11 @@
             <version>3.1.0</version>
         </dependency>
         <dependency>
+            <groupId>org.redisson</groupId>
+            <artifactId>redisson</artifactId>
+            <version>3.36.0</version>
+        </dependency>
+        <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-pool2</artifactId>
             <version>2.6.2</version>

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/controller/BytedanceController.java

@@ -305,8 +305,7 @@ public class BytedanceController {
         for (JSONObject jsonObject : costAccount) {
             executor10Service1.submit(() -> {
                 Long adverId = jsonObject.getLong("advertiserId");
-                String accessToken = jsonObject.getString("accessToken");
-                bytedanceMaterialReportService.getBytedanceMaterialVideo(adverId, null, accessToken, null, null, 1, null, 1, 0,1);
+                bytedanceMaterialReportService.getBytedanceMaterialVideo(adverId, null, null, null, 1, null, 1, 0,1);
             });
         }
     }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/controller/TestController.java

@@ -1,7 +1,9 @@
 package cn.com.ctop.job.bytedance.controller;
 
+import cn.com.ctop.job.bytedance.utils.RateLimitUtil;
 import cn.com.ctop.job.bytedance.utils.RedisUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RateIntervalUnit;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -9,11 +11,15 @@ import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 @RestController
 @RequestMapping("/test")
 @Slf4j
 public class TestController {
+    private static final ExecutorService executorService = Executors.newFixedThreadPool(10);
+
     @Autowired
     private RedisTemplate redisTemplateDefault;
 
@@ -23,6 +29,9 @@ public class TestController {
     @Autowired
     private RedisUtil redisUtils;
 
+    @Autowired
+    private RateLimitUtil rateLimitUtil;
+
     @GetMapping("/test")
     public void test(String key) {
         Object o = redisUtils.get(key);
@@ -41,4 +50,23 @@ public class TestController {
         System.out.println(Objects.requireNonNull(redis2Template_db8.opsForValue().get(key)));
     }
 
+
+    @GetMapping("/testRateLimit")
+    public void testRateLimit(String key) {
+        for (int i = 0; i < 1000; i++) {
+            int finalI = i;
+            executorService.submit(new Runnable() {
+                @Override
+                public void run() {
+                    if(rateLimitUtil.acquire("test1","http://a.com/b/c", 25, 1, RateIntervalUnit.SECONDS)){
+                        log.info(Thread.currentThread().getName() + " : " + finalI);
+                    }else{
+                        log.info(Thread.currentThread().getName() + " : 限流了！！！");
+                    }
+
+                }
+            });
+        }
+    }
+
 }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/mapper/BytedanceRefreshTokenMapper.java

@@ -16,6 +16,8 @@ public interface BytedanceRefreshTokenMapper {
 
     String onlyGetToken(@Param("advertiserId") Long advertiserId);
 
+    JSONObject getAuthConfig(@Param("advertiserId") Long advertiserId);
+
     String onlyGetADQToken(@Param("advertiserId") Long advertiserId);
 
     List<JSONObject> queryOfficialAccounts();

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/mapper/xml/BytedanceRefreshTokenMapper.xml

@@ -61,6 +61,16 @@
         order by id
     </select>
 
+    <select id="getAuthConfig" resultType="com.alibaba.fastjson.JSONObject">
+        SELECT access_token,app_id
+        FROM deepthink.bytedance_auth_list
+        WHERE admin_advertiser_id =
+              (SELECT admin_advertiser_id
+               FROM deepthink.bytedance_advertiser_list
+               WHERE advertiser_id = #{advertiserId}
+            LIMIT 1)
+    </select>
+
 
     <insert id="refreshAppletTokens">
         replace into thirtpart.applet_tokens (

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/IBytedanceMaterialReportService.java

@@ -6,5 +6,5 @@ public interface IBytedanceMaterialReportService {
     void getBytedanceMaterialDailyReport(Long advertiserId, String accessToken, String startDate, String endDate, Integer page,
                                          Long errorId, Integer isLog, Integer isSend, int retry);
 
-    void getBytedanceMaterialVideo(Long advertiserId, Set<Long> materialIds, String accessToken, String startTime, String endTime, Integer page, Long errorId, Integer isLog, Integer isSend, int retry);
+    void getBytedanceMaterialVideo(Long advertiserId, Set<Long> materialIds, String startTime, String endTime, Integer page, Long errorId, Integer isLog, Integer isSend, int retry);
 }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/impl/BytedanceMaterialReportServiceImpl.java

@@ -8,10 +8,12 @@ import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
 import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.FieldsConfig;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
+import cn.com.ctop.job.bytedance.utils.RateLimitUtil;
 import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RateIntervalUnit;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
@@ -33,6 +35,8 @@ public class BytedanceMaterialReportServiceImpl implements IBytedanceMaterialRep
     private BytedanceRefreshTokenMapper tokenMapper;
     @Autowired
     private RedisTemplate redisTemplate_db2;
+    @Autowired
+    private RateLimitUtil rateLimitUtil;
 
     @Override
     public void getBytedanceMaterialDailyReport(Long advertiserId, String accessToken, String startDate, String endDate, Integer page, Long errorId, Integer isLog, Integer isSend, int retry) {
@@ -120,7 +124,7 @@ public class BytedanceMaterialReportServiceImpl implements IBytedanceMaterialRep
             }
 
             if (!Check.isNull(materialIds)) {
-                getBytedanceMaterialVideo(advertiserId, materialIds, accessToken, null, null, 1, null, 1, null, 1);
+                getBytedanceMaterialVideo(advertiserId, materialIds, null, null, 1, null, 1, null, 1);
             }
 
             if (page < totalPage) {
@@ -140,9 +144,15 @@ public class BytedanceMaterialReportServiceImpl implements IBytedanceMaterialRep
         }
     }
 
-    public void getBytedanceMaterialVideo(Long advertiserId, Set<Long> materialIds, String accessToken, String startTime, String endTime, Integer page, Long errorId, Integer isLog, Integer isSend, int retry) {
+    public void getBytedanceMaterialVideo(Long advertiserId, Set<Long> materialIds, String startTime, String endTime, Integer page, Long errorId, Integer isLog, Integer isSend, int retry) {
         try {
-            accessToken = tokenMapper.onlyGetToken(advertiserId);
+            JSONObject authConfig = tokenMapper.getAuthConfig(advertiserId);
+            if (Check.isNull(authConfig) || authConfig.getString("access_token") == null) {
+                log.info("账户id:{}，获取视频素材token为空", advertiserId);
+            }
+            String accessToken = authConfig.getString("access_token");
+            String devAppId = authConfig.getString("app_id");
+
             String url = "https://api.oceanengine.com/open_api/2/file/video/get/";
             Map<String, Object> paramsMap = new HashMap<>();
             paramsMap.put("advertiser_id", advertiserId);
@@ -160,7 +170,10 @@ public class BytedanceMaterialReportServiceImpl implements IBytedanceMaterialRep
             Map<String, String> headers = new HashMap<>();
             headers.put("Access-Token", accessToken);
             headers.put("Content-Type", "application/json");
-            String s = HttpUtils.httpGet(url, paramsMap, headers);
+            String s = null;
+            if (rateLimitUtil.acquire(devAppId, url, 10, 1, RateIntervalUnit.SECONDS)) {
+                s = HttpUtils.httpGet(url, paramsMap, headers);
+            }
             JSONObject result = JSONObject.parseObject(s);
             if (Check.isNull(result)) {
                 log.info("账户id:{}，获取视频素材结果返回为空", advertiserId);
@@ -199,7 +212,7 @@ public class BytedanceMaterialReportServiceImpl implements IBytedanceMaterialRep
                         Thread.sleep(800);
                     } catch (Exception ignored) {
                     }
-                    getBytedanceMaterialVideo(advertiserId, materialIds, accessToken, startTime, endTime, page, errorId, isLog, isSend, retry + 1);
+                    getBytedanceMaterialVideo(advertiserId, materialIds, startTime, endTime, page, errorId, isLog, isSend, retry + 1);
                     return;
                 }
                 log.info("账户id:{}，日期:{}获取视频素材失败，返回数据:{}", advertiserId, startTime, result);
@@ -216,7 +229,7 @@ public class BytedanceMaterialReportServiceImpl implements IBytedanceMaterialRep
                 return;
             }
             if (page < totalPage) {
-                getBytedanceMaterialVideo(advertiserId, materialIds, accessToken, startTime, endTime, page + 1, errorId, isLog, isSend, 1);
+                getBytedanceMaterialVideo(advertiserId, materialIds, startTime, endTime, page + 1, errorId, isLog, isSend, 1);
             } else {
                 log.info("账户ID：{}, 开始日期:{}, 结束日期:{}, totalPage:{}, 获取视频素材成功", advertiserId, startTime, endTime, totalPage);
                 if (errorId != null) {

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/BytedancePromoterReportJob.java

@@ -157,8 +157,7 @@ public class BytedancePromoterReportJob {
         for (JSONObject jsonObject : costAccount) {
             dailyExecutorService.submit(() -> {
                 Long adverId = jsonObject.getLong("advertiserId");
-                String accessToken = jsonObject.getString("accessToken");
-                bytedanceMaterialReportService.getBytedanceMaterialVideo(adverId, null, accessToken, date, nowDate, 1, null, 1, 0, 1);
+                bytedanceMaterialReportService.getBytedanceMaterialVideo(adverId, null, date, nowDate, 1, null, 1, 0, 1);
             });
         }
     }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/BytedanceReportRetryJob.java

@@ -178,7 +178,7 @@ public class BytedanceReportRetryJob {
                         @Override
                         public void run() {
                             if (!Check.isNull(accessToken)) {
-                                bytedanceMaterialReportService.getBytedanceMaterialVideo(advertiserId, materialIds, accessToken, startDate, endDate, 1, id, null, isSend, 1);
+                                bytedanceMaterialReportService.getBytedanceMaterialVideo(advertiserId, materialIds, startDate, endDate, 1, id, null, isSend, 1);
                             }
                         }
                     });

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/RateLimitUtil.java

@@ -0,0 +1,44 @@
+package cn.com.ctop.job.bytedance.utils;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RRateLimiter;
+import org.redisson.api.RateIntervalUnit;
+import org.redisson.api.RateType;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class RateLimitUtil {
+    @Autowired
+    public RedissonClient redissonClient;
+    private static final Map<String, Boolean> initMap = new ConcurrentHashMap<>();
+
+    /**
+     * 自定义限流规则
+     */
+    public boolean acquire(String devAppId, String url, long limit, long interval, RateIntervalUnit unit) {
+        URI uri = URI.create(url);
+        String limitKey = uri.getPath().replaceAll("/$", "").replace("/", ":");
+        String key = "rate:limit" + limitKey + ":" + devAppId;
+        if (!initMap.containsKey(key)) {
+            RRateLimiter limiter = redissonClient.getRateLimiter(key);
+            limiter.trySetRate(
+                    RateType.OVERALL,
+                    limit,
+                    interval,
+                    unit
+            );
+            initMap.put(key, true);
+        }
+        return redissonClient.getRateLimiter(key).tryAcquire(1, 2, TimeUnit.SECONDS);
+    }
+}

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/RedisConfig.java

@@ -1,5 +1,8 @@
 package cn.com.ctop.job.bytedance.utils;
 
+import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.redisson.config.Config;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -23,6 +26,13 @@ public class RedisConfig {
     @Value("${spring.redis.password}")
     private String password;
 
+    @Bean
+    public RedissonClient redissonClient() {
+        Config config = new Config();
+        config.useSingleServer().setAddress("redis://" + host + ":" + port).setDatabase(database).setPassword(password);
+        return Redisson.create(config);
+    }
+
     @Bean(name = "redisTemplateDefault")
     public RedisTemplate<String, Object> redisTemplateDefault() {
         return createTemplate(database);