2929public class AccountGuardAggregator {
3030
3131 private static final int DEFAULT_TTL_SECONDS = 30 ;
32+ private static final int DEFAULT_CONSECUTIVE_TIMES = 2 ;
3233
3334 private static final class NodeStat {
3435 final int connections ;
@@ -43,6 +44,8 @@ private static final class NodeStat {
4344 }
4445
4546 private final Map <String , Map <String , NodeStat >> table = new ConcurrentHashMap <>();
47+ private final Map <String , Integer > consecutiveExceedCounts = new ConcurrentHashMap <>();
48+ private final Map <String , GuardBlockedEntry > blockedAccounts = new ConcurrentHashMap <>();
4649
4750 private final AccountRepository accountRepository ;
4851 private final SystemConfigService systemConfigService ;
@@ -111,7 +114,7 @@ public AccountGuardEvaluation reportAndEvaluateWithStats(GuardSyncRequest reques
111114 if (!reportedAccountNos .contains (accountNo )) {
112115 continue ;
113116 }
114- GuardBlockedEntry entry = evaluateAccount ( account , stats );
117+ GuardBlockedEntry entry = blockedAccounts . get ( accountNo );
115118 if (entry != null ) {
116119 blocked .put (accountNo , entry );
117120 }
@@ -120,6 +123,53 @@ public AccountGuardEvaluation reportAndEvaluateWithStats(GuardSyncRequest reques
120123 return new AccountGuardEvaluation (blocked , statsByAccountNo );
121124 }
122125
126+ /**
127+ * 按中心统一采样周期评估所有已跟踪账户。只有连续超限达到配置次数后，
128+ * 才将账户加入黑名单；任一采样周期恢复到限制内时立即清零并移出黑名单。
129+ */
130+ public AccountGuardEvaluation evaluateTrackedAccounts () {
131+ Set <String > accountNos = new HashSet <>(table .keySet ());
132+ accountNos .addAll (consecutiveExceedCounts .keySet ());
133+ accountNos .addAll (blockedAccounts .keySet ());
134+ if (accountNos .isEmpty ()) {
135+ return new AccountGuardEvaluation (Map .of (), Map .of ());
136+ }
137+
138+ long now = nowEpochSeconds ();
139+ int ttl = getTtlSeconds ();
140+ int requiredTimes = getConsecutiveTimesThreshold ();
141+ Map <String , Account > accountMap = loadAccounts (accountNos );
142+ Map <String , AccountGuardStats > statsByAccountNo = new HashMap <>();
143+
144+ for (String accountNo : accountNos ) {
145+ Account account = accountMap .get (accountNo );
146+ if (account == null ) {
147+ consecutiveExceedCounts .remove (accountNo );
148+ blockedAccounts .remove (accountNo );
149+ continue ;
150+ }
151+
152+ AccountGuardStats stats = calculateStats (accountNo , now , ttl );
153+ statsByAccountNo .put (accountNo , stats );
154+ GuardBlockedEntry exceeded = evaluateAccount (account , stats );
155+ if (exceeded == null ) {
156+ consecutiveExceedCounts .remove (accountNo );
157+ blockedAccounts .remove (accountNo );
158+ continue ;
159+ }
160+
161+ int consecutiveTimes = consecutiveExceedCounts .merge (
162+ accountNo ,
163+ 1 ,
164+ (current , increment ) -> Math .min (requiredTimes , current + increment ));
165+ if (consecutiveTimes >= requiredTimes ) {
166+ blockedAccounts .put (accountNo , exceeded );
167+ }
168+ }
169+
170+ return new AccountGuardEvaluation (new HashMap <>(blockedAccounts ), statsByAccountNo );
171+ }
172+
123173 private AccountGuardStats calculateStats (String accountNo , long now , int ttl ) {
124174 Map <String , NodeStat > byNode = table .get (accountNo );
125175 if (byNode == null || byNode .isEmpty ()) {
@@ -205,6 +255,11 @@ public int getTtlSeconds() {
205255 "airopscat.account.guard.ttl-seconds" , DEFAULT_TTL_SECONDS ));
206256 }
207257
258+ private int getConsecutiveTimesThreshold () {
259+ return Math .max (1 , systemConfigService .getIntValue (
260+ "airopscat.account.connection-limit.alert.consecutive-times" , DEFAULT_CONSECUTIVE_TIMES ));
261+ }
262+
208263 private long nowEpochSeconds () {
209264 return java .time .Instant .now ().getEpochSecond ();
210265 }
0 commit comments