调整限流缓存时间

example/src/main/java/com/example/demo/TestExistsTransactionalRetryService.java

@@ -29,7 +29,7 @@ public class TestExistsTransactionalRetryService {
     @Autowired
     private RemoteService remoteService;
 
-    @Retryable(scene = "testSimpleInsert", bizNo = "#name", localTimes = 5)
+    @Retryable(scene = "testSimpleInsert", bizNo = "#name", localTimes = 3)
     @Transactional
     public String testSimpleInsert(String name) {
 

example/src/test/java/com/example/ExistsTransactionalRetryServiceTest.java

@@ -79,7 +79,7 @@ public class ExistsTransactionalRetryServiceTest {
     @SneakyThrows
     @Test
     public void syncTestSimpleInsert() {
-        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
+        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 50, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
 
         Mockito.when(remoteService.call())
                 .thenReturn(new Result(0, "1"))
@@ -89,7 +89,7 @@ public class ExistsTransactionalRetryServiceTest {
                 .thenReturn(new Result(0, "5"))
         ;
         try {
-            for (int i = 0; i < 100; i++) {
+            for (int i = 0; i < 300; i++) {
                 threadPoolExecutor.execute(() -> testExistsTransactionalRetryService.testSimpleInsert(UUID.randomUUID().toString()));
             }
         } catch (Exception e) {

x-retry-server/Dockerfile

@@ -1,5 +1,6 @@
 FROM openjdk:8-jdk-alpine
 MAINTAINER www.byteblogs.com
+LABEL server-name=x-retry-server
 
 ADD ./target/x-retry-server.jar x-retry-server.jar
 

x-retry-server/src/main/java/com/x/retry/server/support/cache/CacheGroupRateLimiter.java

@@ -47,8 +47,6 @@ public class CacheGroupRateLimiter implements Lifecycle {
         CACHE = CacheBuilder.newBuilder()
                 // 设置并发级别为cpu核心数
                 .concurrencyLevel(Runtime.getRuntime().availableProcessors())
-                .expireAfterWrite(10, TimeUnit.SECONDS)
-                .expireAfterAccess(10, TimeUnit.SECONDS)
                 .build();
     }
 

x-retry-server/src/main/java/com/x/retry/server/support/dispatch/DispatchService.java

@@ -124,12 +124,10 @@ public class DispatchService implements Lifecycle {
         for (ServerNode serverNode : serverNodes) {
             RateLimiter rateLimiter = rateLimiterCache.getIfPresent(serverNode.getHostId());
             if (Objects.isNull(rateLimiter)) {
-                rateLimiterCache.put(groupName, RateLimiter.create(systemProperties.getLimiter()));
+                rateLimiterCache.put(serverNode.getHostId(), RateLimiter.create(systemProperties.getLimiter()));
             }
         }
 
-        rateLimiterCache.invalidateAll();
-
     }
 
     /**

x-retry-server/src/main/java/com/x/retry/server/support/dispatch/actor/exec/ExecUnitActor.java

@@ -98,7 +98,6 @@ public class ExecUnitActor extends AbstractActor  {
                 retryTaskLog.setId(null);
                 Assert.isTrue(1 ==  retryTaskLogMapper.insert(retryTaskLog),
                         new XRetryServerException("新增重试日志失败"));
-
             }
 
         }).build();

x-retry-server/src/main/java/com/x/retry/server/support/strategy/FilterStrategies.java

@@ -175,7 +175,7 @@ public class FilterStrategies {
             ServerNode serverNode = context.getServerNode();
 
             RateLimiter rateLimiter = CacheGroupRateLimiter.getRateLimiterByKey(serverNode.getHostId());
-            if (!rateLimiter.tryAcquire(1, TimeUnit.SECONDS)) {
+            if (Objects.nonNull(rateLimiter) && !rateLimiter.tryAcquire(100, TimeUnit.MILLISECONDS)) {
                 LogUtils.error("该POD:[{}]已到达最大限流阈值，本次重试不执行", serverNode.getHostId());
                 return Boolean.FALSE;
             }

x-retry-server/src/main/resources/application.yml

@@ -35,5 +35,6 @@ x-retry:
   retryPullPageSize: 100 # 拉取重试数据的每批次的大小
   nettyPort: 1788  # 服务端netty端口
   totalPartition: 32  # 重试和死信表的分区总数
+  limiter: 10