diff --git a/external/external-core/src/main/java/com/iohao/game/external/core/config/ExternalGlobalConfig.java b/external/external-core/src/main/java/com/iohao/game/external/core/config/ExternalGlobalConfig.java index 275537d75..e2d1bd22f 100644 --- a/external/external-core/src/main/java/com/iohao/game/external/core/config/ExternalGlobalConfig.java +++ b/external/external-core/src/main/java/com/iohao/game/external/core/config/ExternalGlobalConfig.java @@ -56,4 +56,43 @@ public class CoreOption { /** http 升级 websocket 协议地址 */ public String websocketPath = "/websocket"; } + + /** + * 流量过载保护配置 + */ + @UtilityClass + public class TrafficProtectionOption { + /** 是否启用流量保护功能 */ + public boolean enableTrafficProtection = false; + + /** 最大连接数限制，默认5000 */ + public int maxConnections = 5000; + + /** 是否启用自适应限流 */ + public boolean enableAdaptiveRateLimit = true; + + /** 监控间隔（毫秒），默认1000ms */ + public int monitorInterval = 1000; + + /** 自适应限流窗口大小（秒），默认60秒 */ + public int adaptiveWindowSize = 60; + + /** 最小限流阈值（连接数），默认1000 */ + public int minRateLimitThreshold = 1000; + + /** 最大限流阈值（连接数），默认8000 */ + public int maxRateLimitThreshold = 8000; + + /** CPU使用率阈值（百分比），超过此值将触发限流 */ + public double cpuThreshold = 80.0; + + /** 内存使用率阈值（百分比），超过此值将触发限流 */ + public double memoryThreshold = 85.0; + + /** 是否启用连接拒绝日志 */ + public boolean enableRejectionLog = true; + + /** 是否启用系统资源监控 */ + public boolean enableResourceMonitoring = true; + } } diff --git a/external/external-core/src/main/java/com/iohao/game/external/core/monitor/SystemResourceMonitor.java b/external/external-core/src/main/java/com/iohao/game/external/core/monitor/SystemResourceMonitor.java new file mode 100644 index 000000000..b0031e049 --- /dev/null +++ b/external/external-core/src/main/java/com/iohao/game/external/core/monitor/SystemResourceMonitor.java @@ -0,0 +1,414 @@ +/* + * ioGame + * Copyright (C) 2021 - present 渔民小镇 （262610965@qq.com、luoyizhu@gmail.com） . All Rights Reserved. + * # iohao.com . 渔民小镇 + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package com.iohao.game.external.core.monitor; + +import com.iohao.game.external.core.config.ExternalGlobalConfig; +import lombok.Getter; +import lombok.experimental.Accessors; +import lombok.extern.slf4j.Slf4j; + +import java.lang.management.ManagementFactory; +import java.lang.management.MemoryMXBean; +import java.lang.management.OperatingSystemMXBean; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +/** + * 系统资源监控器 + * 负责监控CPU、内存等系统资源使用情况 + * + * @author undertaker86001 + * @date 2025-08-15 + */ +@Slf4j +public class SystemResourceMonitor { + + /** 单例实例 */ + private static final SystemResourceMonitor INSTANCE = new SystemResourceMonitor(); + + /** 操作系统MXBean */ + private final OperatingSystemMXBean osBean = ManagementFactory.getOperatingSystemMXBean(); + + /** 内存MXBean */ + private final MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean(); + + /** 定时任务执行器 */ + private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); + + /** 当前系统资源信息 */ + private final AtomicReference currentInfo = new AtomicReference<>(); + + /** 是否已启动 */ + private volatile boolean started = false; + + /** 私有构造函数 */ + private SystemResourceMonitor() { + // 初始化当前资源信息 + this.currentInfo.set(new SystemResourceInfo()); + } + + /** + * 获取单例实例 + * + * @return SystemResourceMonitor实例 + */ + public static SystemResourceMonitor getInstance() { + return INSTANCE; + } + + /** + * 启动监控 + */ + public void start() { + if (started) { + return; + } + + if (!ExternalGlobalConfig.TrafficProtectionOption.enableResourceMonitoring) { + log.info("系统资源监控已禁用"); + return; + } + + // 立即执行一次监控 + updateResourceInfo(); + + // 启动定时监控任务 + int interval = ExternalGlobalConfig.TrafficProtectionOption.monitorInterval; + scheduler.scheduleAtFixedRate(this::updateResourceInfo, interval, interval, TimeUnit.MILLISECONDS); + + started = true; + log.info("系统资源监控已启动，监控间隔: {}ms", interval); + } + + /** + * 停止监控 + */ + public void stop() { + if (!started) { + return; + } + + scheduler.shutdown(); + try { + if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) { + scheduler.shutdownNow(); + } + } catch (InterruptedException e) { + scheduler.shutdownNow(); + Thread.currentThread().interrupt(); + } + + started = false; + log.info("系统资源监控已停止"); + } + + /** + * 获取当前系统资源信息 + * + * @return 系统资源信息 + */ + public SystemResourceInfo getCurrentInfo() { + return currentInfo.get(); + } + + /** + * 更新系统资源信息 + */ + private void updateResourceInfo() { + try { + SystemResourceInfo info = new SystemResourceInfo(); + + // 获取CPU使用率 + info.setCpuUsage(getCpuUsage()); + + // 获取内存使用率 + info.setMemoryUsage(getMemoryUsage()); + + // 获取系统负载 + info.setSystemLoad(getSystemLoad()); + + // 获取可用内存 + info.setAvailableMemory(getAvailableMemory()); + + // 获取总内存 + info.setTotalMemory(getTotalMemory()); + + // 获取JVM内存使用情况 + info.setJvmMemoryUsage(getJvmMemoryUsage()); + + // 设置时间戳 + info.setTimestamp(System.currentTimeMillis()); + + // 更新当前信息 + currentInfo.set(info); + + // 检查是否需要告警 + checkResourceAlerts(info); + + } catch (Exception e) { + log.error("更新系统资源信息失败", e); + } + } + + /** + * 获取CPU使用率 + * + * @return CPU使用率（百分比） + */ + private double getCpuUsage() { + try { + // 获取系统CPU负载 + double systemLoad = osBean.getSystemLoadAverage(); + + // 如果系统负载不可用，返回0 + if (systemLoad < 0) { + return 0.0; + } + + // 获取CPU核心数 + int cpuCores = osBean.getAvailableProcessors(); + + // 计算CPU使用率（系统负载 / CPU核心数 * 100） + double cpuUsage = (systemLoad / cpuCores) * 100.0; + + // 限制在0-100范围内 + return Math.min(100.0, Math.max(0.0, cpuUsage)); + + } catch (Exception e) { + log.warn("获取CPU使用率失败", e); + return 0.0; + } + } + + /** + * 获取内存使用率 + * + * @return 内存使用率（百分比） + */ + private double getMemoryUsage() { + try { + long totalMemory = getTotalMemory(); + long availableMemory = getAvailableMemory(); + + if (totalMemory <= 0) { + return 0.0; + } + + long usedMemory = totalMemory - availableMemory; + double memoryUsage = ((double) usedMemory / totalMemory) * 100.0; + + return Math.min(100.0, Math.max(0.0, memoryUsage)); + + } catch (Exception e) { + log.warn("获取内存使用率失败", e); + return 0.0; + } + } + + /** + * 获取系统负载 + * + * @return 系统负载 + */ + private double getSystemLoad() { + try { + double load = osBean.getSystemLoadAverage(); + return load < 0 ? 0.0 : load; + } catch (Exception e) { + log.warn("获取系统负载失败", e); + return 0.0; + } + } + + /** + * 获取可用内存 + * + * @return 可用内存（字节） + */ + private long getAvailableMemory() { + try { + return Runtime.getRuntime().freeMemory(); + } catch (Exception e) { + log.warn("获取可用内存失败", e); + return 0L; + } + } + + /** + * 获取总内存 + * + * @return 总内存（字节） + */ + private long getTotalMemory() { + try { + return Runtime.getRuntime().totalMemory(); + } catch (Exception e) { + log.warn("获取总内存失败", e); + return 0L; + } + } + + /** + * 获取JVM内存使用率 + * + * @return JVM内存使用率（百分比） + */ + private double getJvmMemoryUsage() { + try { + long usedMemory = memoryBean.getHeapMemoryUsage().getUsed(); + long maxMemory = memoryBean.getHeapMemoryUsage().getMax(); + + if (maxMemory <= 0) { + return 0.0; + } + + double jvmMemoryUsage = ((double) usedMemory / maxMemory) * 100.0; + return Math.min(100.0, Math.max(0.0, jvmMemoryUsage)); + + } catch (Exception e) { + log.warn("获取JVM内存使用率失败", e); + return 0.0; + } + } + + /** + * 检查资源告警 + * + * @param info 系统资源信息 + */ + private void checkResourceAlerts(SystemResourceInfo info) { + double cpuThreshold = ExternalGlobalConfig.TrafficProtectionOption.cpuThreshold; + double memoryThreshold = ExternalGlobalConfig.TrafficProtectionOption.memoryThreshold; + + // CPU使用率告警 + if (info.getCpuUsage() > cpuThreshold) { + log.warn("CPU使用率过高: {}%, 阈值: {}%", + String.format("%.2f", info.getCpuUsage()), + String.format("%.2f", cpuThreshold)); + } + + // 内存使用率告警 + if (info.getMemoryUsage() > memoryThreshold) { + log.warn("内存使用率过高: {}%, 阈值: {}%", + String.format("%.2f", info.getMemoryUsage()), + String.format("%.2f", memoryThreshold)); + } + + // JVM内存使用率告警 + if (info.getJvmMemoryUsage() > 90.0) { + log.warn("JVM内存使用率过高: {}%", + String.format("%.2f", info.getJvmMemoryUsage())); + } + } + + /** + * 系统资源信息 + */ + @Getter + @Accessors(chain = true) + public static class SystemResourceInfo { + /** CPU使用率（百分比） */ + private double cpuUsage = 0.0; + + /** 内存使用率（百分比） */ + private double memoryUsage = 0.0; + + /** 系统负载 */ + private double systemLoad = 0.0; + + /** 可用内存（字节） */ + private long availableMemory = 0L; + + /** 总内存（字节） */ + private long totalMemory = 0L; + + /** JVM内存使用率（百分比） */ + private double jvmMemoryUsage = 0.0; + + /** 时间戳 */ + private long timestamp = 0L; + + /** + * 设置CPU使用率 + */ + public SystemResourceInfo setCpuUsage(double cpuUsage) { + this.cpuUsage = cpuUsage; + return this; + } + + /** + * 设置内存使用率 + */ + public SystemResourceInfo setMemoryUsage(double memoryUsage) { + this.memoryUsage = memoryUsage; + return this; + } + + /** + * 设置系统负载 + */ + public SystemResourceInfo setSystemLoad(double systemLoad) { + this.systemLoad = systemLoad; + return this; + } + + /** + * 设置可用内存 + */ + public SystemResourceInfo setAvailableMemory(long availableMemory) { + this.availableMemory = availableMemory; + return this; + } + + /** + * 设置总内存 + */ + public SystemResourceInfo setTotalMemory(long totalMemory) { + this.totalMemory = totalMemory; + return this; + } + + /** + * 设置JVM内存使用率 + */ + public SystemResourceInfo setJvmMemoryUsage(double jvmMemoryUsage) { + this.jvmMemoryUsage = jvmMemoryUsage; + return this; + } + + /** + * 设置时间戳 + */ + public SystemResourceInfo setTimestamp(long timestamp) { + this.timestamp = timestamp; + return this; + } + + @Override + public String toString() { + return String.format( + "SystemResourceInfo{cpuUsage=%.2f%%, memoryUsage=%.2f%%, systemLoad=%.2f, " + + "availableMemory=%d, totalMemory=%d, jvmMemoryUsage=%.2f%%, timestamp=%d}", + cpuUsage, memoryUsage, systemLoad, availableMemory, totalMemory, jvmMemoryUsage, timestamp + ); + } + } +} diff --git a/external/external-core/src/main/java/com/iohao/game/external/core/protection/TrafficProtector.java b/external/external-core/src/main/java/com/iohao/game/external/core/protection/TrafficProtector.java new file mode 100644 index 000000000..3740f0469 --- /dev/null +++ b/external/external-core/src/main/java/com/iohao/game/external/core/protection/TrafficProtector.java @@ -0,0 +1,184 @@ +package com.iohao.game.external.core.protection; + +import com.iohao.game.external.core.config.ExternalGlobalConfig; +import com.iohao.game.external.core.ratelimit.AdaptiveRateLimiter; +import lombok.extern.slf4j.Slf4j; + +/** + * 流量保护器 + * 负责在连接建立时进行流量控制和保护 + * @author undertaker86001 + * @date 2025-08-15 + */ +@Slf4j +public class TrafficProtector { + + private static final TrafficProtector INSTANCE = new TrafficProtector(); + private final AdaptiveRateLimiter rateLimiter = AdaptiveRateLimiter.getInstance(); + + private TrafficProtector() {} + + public static TrafficProtector getInstance() { + return INSTANCE; + } + + /** + * 检查是否允许新连接 + * + * @param clientIp 客户端IP地址 + * @return 连接检查结果 + */ + public ConnectionCheckResult checkConnection(String clientIp) { + // 检查是否启用流量保护 + if (!ExternalGlobalConfig.TrafficProtectionOption.enableTrafficProtection) { + return ConnectionCheckResult.allowed(); + } + + try { + // 尝试获取连接许可 + if (rateLimiter.tryAcquire()) { + // 允许连接，增加连接数 + rateLimiter.incrementConnections(); + + if (ExternalGlobalConfig.TrafficProtectionOption.enableRejectionLog) { + log.info("连接允许 - IP: {}, 当前连接数: {}, 阈值: {}", + clientIp, rateLimiter.getCurrentConnections(), rateLimiter.getCurrentThreshold()); + } + + return ConnectionCheckResult.allowed(); + } else { + // 拒绝连接 + if (ExternalGlobalConfig.TrafficProtectionOption.enableRejectionLog) { + log.warn("连接拒绝 - IP: {}, 当前连接数: {}, 阈值: {}, 原因: 连接数超限", + clientIp, rateLimiter.getCurrentConnections(), rateLimiter.getCurrentThreshold()); + } + + return ConnectionCheckResult.rejected(ConnectionRejectionReason.CONNECTION_LIMIT_EXCEEDED); + } + + } catch (Exception e) { + log.error("流量保护检查失败 - IP: {}", clientIp, e); + + // 发生异常时，为了安全起见，拒绝连接 + return ConnectionCheckResult.rejected(ConnectionRejectionReason.SYSTEM_ERROR); + } + } + + /** + * 连接断开时减少连接数 + */ + public void onConnectionClosed() { + if (ExternalGlobalConfig.TrafficProtectionOption.enableTrafficProtection) { + rateLimiter.decrementConnections(); + } + } + + /** + * 获取当前连接数 + * + * @return 当前连接数 + */ + public int getCurrentConnections() { + return rateLimiter.getCurrentConnections(); + } + + /** + * 获取当前限流阈值 + * + * @return 当前限流阈值 + */ + public int getCurrentThreshold() { + return rateLimiter.getCurrentThreshold(); + } + + /** + * 获取限流统计信息 + * + * @return 限流统计信息 + */ + public AdaptiveRateLimiter.RateLimitStats getRateLimitStats() { + return rateLimiter.getStats(); + } + + /** + * 启动流量保护器 + */ + public void start() { + if (ExternalGlobalConfig.TrafficProtectionOption.enableTrafficProtection) { + rateLimiter.start(); + log.info("流量保护器已启动"); + } + } + + /** + * 停止流量保护器 + */ + public void stop() { + if (ExternalGlobalConfig.TrafficProtectionOption.enableTrafficProtection) { + rateLimiter.stop(); + log.info("流量保护器已停止"); + } + } + + /** + * 连接检查结果 + */ + public static class ConnectionCheckResult { + private final boolean allowed; + private final ConnectionRejectionReason rejectionReason; + private final String errorMessage; + + private ConnectionCheckResult(boolean allowed, ConnectionRejectionReason rejectionReason, String errorMessage) { + this.allowed = allowed; + this.rejectionReason = rejectionReason; + this.errorMessage = errorMessage; + } + + public static ConnectionCheckResult allowed() { + return new ConnectionCheckResult(true, null, null); + } + + public static ConnectionCheckResult rejected(ConnectionRejectionReason reason) { + return new ConnectionCheckResult(false, reason, reason.getMessage()); + } + + public boolean isAllowed() { + return allowed; + } + + public ConnectionRejectionReason getRejectionReason() { + return rejectionReason; + } + + public String getErrorMessage() { + return errorMessage; + } + } + + /** + * 连接拒绝原因 + */ + public enum ConnectionRejectionReason { + CONNECTION_LIMIT_EXCEEDED(1001, "连接数超限"), + SYSTEM_OVERLOAD(1002, "系统负载过高"), + RESOURCE_INSUFFICIENT(1003, "资源不足"), + RATE_LIMIT_TRIGGERED(1004, "限流保护触发"), + SYSTEM_ERROR(1005, "系统错误"); + + private final int errorCode; + private final String message; + + ConnectionRejectionReason(int errorCode, String message) { + this.errorCode = errorCode; + this.message = message; + } + + public int getErrorCode() { + return errorCode; + } + + public String getMessage() { + return message; + } + } +} diff --git a/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/AbstractRateLimiter.java b/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/AbstractRateLimiter.java new file mode 100644 index 000000000..a0f8b37f1 --- /dev/null +++ b/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/AbstractRateLimiter.java @@ -0,0 +1,272 @@ +package com.iohao.game.external.core.ratelimit; + +import com.iohao.game.external.core.config.ExternalGlobalConfig; +import com.iohao.game.external.core.monitor.SystemResourceMonitor; +import lombok.Getter; +import lombok.experimental.Accessors; +import lombok.extern.slf4j.Slf4j; + +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +/** + * 限流器抽象基类 + * 提供通用的实现逻辑，子类只需要实现具体的限流算法 + * @author undertaker86001 + * @date 2025-08-15 + */ +@Slf4j +public abstract class AbstractRateLimiter implements RateLimiter { + + protected final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); + + /** 当前限流阈值 */ + protected final AtomicInteger currentLimit = new AtomicInteger(); + + /** 当前连接数 */ + protected final AtomicInteger currentConnections = new AtomicInteger(0); + + /** 被拒绝的连接数 */ + protected final AtomicLong rejectedConnections = new AtomicLong(0); + + /** 总连接请求数 */ + protected final AtomicLong totalRequests = new AtomicLong(0); + + /** 限流统计信息 */ + protected final AtomicReference stats = new AtomicReference<>(); + + /** 是否已启动 */ + protected volatile boolean started = false; + + /** 系统资源监控器 */ + protected final SystemResourceMonitor resourceMonitor = SystemResourceMonitor.getInstance(); + + /** 样本统计 */ + protected final AtomicLong sampleCount = new AtomicLong(0); + protected final AtomicLong totalRtt = new AtomicLong(0); + + protected AbstractRateLimiter() { + int maxConnections = ExternalGlobalConfig.TrafficProtectionOption.maxConnections; + this.currentLimit.set(maxConnections); + this.stats.set(new BaseRateLimitStats()); + } + + @Override + public void start() { + if (started) return; + + if (!ExternalGlobalConfig.TrafficProtectionOption.enableAdaptiveRateLimit) { + log.info("自适应限流已禁用"); + return; + } + + resourceMonitor.start(); + adjustLimit(); + + int interval = ExternalGlobalConfig.TrafficProtectionOption.monitorInterval; + scheduler.scheduleAtFixedRate(this::adjustLimit, interval, interval, TimeUnit.MILLISECONDS); + + started = true; + log.info("{} 自适应限流器已启动，监控间隔: {}ms", getAlgorithmName(), interval); + } + + @Override + public void stop() { + if (!started) return; + + scheduler.shutdown(); + try { + if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) { + scheduler.shutdownNow(); + } + } catch (InterruptedException e) { + scheduler.shutdownNow(); + Thread.currentThread().interrupt(); + } + + resourceMonitor.stop(); + started = false; + log.info("{} 自适应限流器已停止", getAlgorithmName()); + } + + @Override + public boolean tryAcquire() { + totalRequests.incrementAndGet(); + + int limit = currentLimit.get(); + int connections = currentConnections.get(); + + if (connections >= limit) { + rejectedConnections.incrementAndGet(); + return false; + } + + return true; + } + + @Override + public void incrementConnections() { + currentConnections.incrementAndGet(); + } + + @Override + public void decrementConnections() { + currentConnections.decrementAndGet(); + } + + @Override + public int getCurrentConnections() { + return currentConnections.get(); + } + + @Override + public int getCurrentLimit() { + return currentLimit.get(); + } + + @Override + public void recordRtt(double rtt) { + if (rtt <= 0) return; + + sampleCount.incrementAndGet(); + totalRtt.addAndGet((long) (rtt * 1000)); // 转换为微秒 + + // 子类实现具体的RTT更新逻辑 + updateRtt(rtt); + } + + @Override + public RateLimitStats getStats() { + BaseRateLimitStats currentStats = stats.get(); + currentStats.setCurrentConnections(currentConnections.get()); + currentStats.setCurrentLimit(currentLimit.get()); + currentStats.setRejectedConnections(rejectedConnections.get()); + currentStats.setTotalRequests(totalRequests.get()); + currentStats.setTimestamp(System.currentTimeMillis()); + + // 子类实现具体的统计信息更新 + updateStats(currentStats); + + return currentStats; + } + + /** + * 获取算法名称 + */ + protected abstract String getAlgorithmName(); + + /** + * 调整限流阈值 + * 子类实现具体的限流算法 + */ + protected abstract void adjustLimit(); + + /** + * 更新RTT值 + * 子类实现具体的RTT更新逻辑 + */ + protected abstract void updateRtt(double rtt); + + /** + * 更新统计信息 + * 子类实现具体的统计信息更新 + */ + protected abstract void updateStats(BaseRateLimitStats stats); + + /** + * 应用系统资源约束 + */ + protected int applyResourceConstraints(int newLimit, SystemResourceMonitor.SystemResourceInfo resourceInfo) { + double cpuUsage = resourceInfo.getCpuUsage(); + double memoryUsage = resourceInfo.getMemoryUsage(); + + double cpuThreshold = ExternalGlobalConfig.TrafficProtectionOption.cpuThreshold; + double memoryThreshold = ExternalGlobalConfig.TrafficProtectionOption.memoryThreshold; + + // 如果CPU或内存使用率过高，减少limit + if (cpuUsage > cpuThreshold || memoryUsage > memoryThreshold) { + double reductionFactor = 0.8; // 减少20% + newLimit = (int) (newLimit * reductionFactor); + } + + return newLimit; + } + + /** + * 基础限流统计信息 + */ + @Getter + @Accessors(chain = true) + protected static class BaseRateLimitStats implements RateLimitStats { + private int currentConnections = 0; + private int currentLimit = 0; + private long rejectedConnections = 0; + private long totalRequests = 0; + private double cpuUsage = 0.0; + private double memoryUsage = 0.0; + private double systemLoad = 0.0; + private double jvmMemoryUsage = 0.0; + private long timestamp = 0L; + + public BaseRateLimitStats setCurrentConnections(int currentConnections) { + this.currentConnections = currentConnections; + return this; + } + + public BaseRateLimitStats setCurrentLimit(int currentLimit) { + this.currentLimit = currentLimit; + return this; + } + + public BaseRateLimitStats setRejectedConnections(long rejectedConnections) { + this.rejectedConnections = rejectedConnections; + return this; + } + + public BaseRateLimitStats setTotalRequests(long totalRequests) { + this.totalRequests = totalRequests; + return this; + } + + public BaseRateLimitStats setCpuUsage(double cpuUsage) { + this.cpuUsage = cpuUsage; + return this; + } + + public BaseRateLimitStats setMemoryUsage(double memoryUsage) { + this.memoryUsage = memoryUsage; + return this; + } + + public BaseRateLimitStats setSystemLoad(double systemLoad) { + this.systemLoad = systemLoad; + return this; + } + + public BaseRateLimitStats setJvmMemoryUsage(double jvmMemoryUsage) { + this.jvmMemoryUsage = jvmMemoryUsage; + return this; + } + + public BaseRateLimitStats setTimestamp(long timestamp) { + this.timestamp = timestamp; + return this; + } + + @Override + public double getRejectionRate() { + if (totalRequests == 0) return 0.0; + return ((double) rejectedConnections / totalRequests) * 100.0; + } + + @Override + public double getConnectionRate() { + if (currentLimit == 0) return 0.0; + return ((double) currentConnections / currentLimit) * 100.0; + } + } +} diff --git a/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/AdaptiveRateLimiter.java b/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/AdaptiveRateLimiter.java new file mode 100644 index 000000000..6490dad11 --- /dev/null +++ b/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/AdaptiveRateLimiter.java @@ -0,0 +1,437 @@ +package com.iohao.game.external.core.ratelimit; + +import com.iohao.game.external.core.config.ExternalGlobalConfig; +import com.iohao.game.external.core.monitor.SystemResourceMonitor; +import lombok.Getter; +import lombok.experimental.Accessors; +import lombok.extern.slf4j.Slf4j; + +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +/** + * 基于Netflix Gradient算法的自适应限流器 + * 通过计算无负载时的RTT与当前RTT的比值来判断是否出现请求排队 + * + * 参考：https://gummary.github.io/post/netfix%E8%87%AA%E9%80%82%E5%BA%94%E9%99%90%E6%B5%81%E7%AE%97%E6%B3%95/ + */ +@Slf4j +public class AdaptiveRateLimiter { + + private static final AdaptiveRateLimiter INSTANCE = new AdaptiveRateLimiter(); + private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); + + /** 当前限流阈值 */ + private final AtomicInteger currentLimit = new AtomicInteger(); + + /** 当前连接数 */ + private final AtomicInteger currentConnections = new AtomicInteger(0); + + /** 被拒绝的连接数 */ + private final AtomicLong rejectedConnections = new AtomicLong(0); + + /** 总连接请求数 */ + private final AtomicLong totalRequests = new AtomicLong(0); + + /** 限流统计信息 */ + private final AtomicReference stats = new AtomicReference<>(); + + /** 是否已启动 */ + private volatile boolean started = false; + + /** 系统资源监控器 */ + private final SystemResourceMonitor resourceMonitor = SystemResourceMonitor.getInstance(); + + /** RTT相关参数 */ + private volatile double rttNoLoad = 0.0; // 无负载时的RTT + private volatile double rttActual = 0.0; // 当前实际RTT + private volatile long lastUpdateTime = 0L; // 上次更新时间 + + /** 自适应参数 */ + private static final double MIN_GRADIENT = 0.5; // 最小梯度值 + private static final double MAX_GRADIENT = 1.0; // 最大梯度值 + private static final double ALPHA = 0.1; // 平滑因子 + private static final int MIN_SAMPLE_SIZE = 10; // 最小样本数 + + /** 样本统计 */ + private final AtomicLong sampleCount = new AtomicLong(0); + private final AtomicLong totalRtt = new AtomicLong(0); + + private AdaptiveRateLimiter() { + int maxConnections = ExternalGlobalConfig.TrafficProtectionOption.maxConnections; + this.currentLimit.set(maxConnections); + this.stats.set(new RateLimitStats()); + + // 初始化RTT值 + this.rttNoLoad = 1.0; // 默认1ms + this.rttActual = 1.0; + } + + public static AdaptiveRateLimiter getInstance() { + return INSTANCE; + } + + public void start() { + if (started) return; + + if (!ExternalGlobalConfig.TrafficProtectionOption.enableAdaptiveRateLimit) { + log.info("自适应限流已禁用"); + return; + } + + resourceMonitor.start(); + adjustLimit(); + + int interval = ExternalGlobalConfig.TrafficProtectionOption.monitorInterval; + scheduler.scheduleAtFixedRate(this::adjustLimit, interval, interval, TimeUnit.MILLISECONDS); + + started = true; + log.info("Gradient自适应限流器已启动，监控间隔: {}ms", interval); + } + + public void stop() { + if (!started) return; + + scheduler.shutdown(); + try { + if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) { + scheduler.shutdownNow(); + } + } catch (InterruptedException e) { + scheduler.shutdownNow(); + Thread.currentThread().interrupt(); + } + + resourceMonitor.stop(); + started = false; + log.info("Gradient自适应限流器已停止"); + } + + public boolean tryAcquire() { + totalRequests.incrementAndGet(); + + int limit = currentLimit.get(); + int connections = currentConnections.get(); + + if (connections >= limit) { + rejectedConnections.incrementAndGet(); + return false; + } + + return true; + } + + public void incrementConnections() { + currentConnections.incrementAndGet(); + } + + public void decrementConnections() { + currentConnections.decrementAndGet(); + } + + /** + * 记录RTT样本 + * + * @param rtt 响应时间（毫秒） + */ + public void recordRtt(double rtt) { + if (rtt <= 0) return; + + sampleCount.incrementAndGet(); + totalRtt.addAndGet((long) (rtt * 1000)); // 转换为微秒 + + // 更新实际RTT（使用指数移动平均） + updateRtt(rtt); + } + + public int getCurrentConnections() { + return currentConnections.get(); + } + + public int getCurrentLimit() { + return currentLimit.get(); + } + + public RateLimitStats getStats() { + RateLimitStats currentStats = stats.get(); + currentStats.setCurrentConnections(currentConnections.get()); + currentStats.setCurrentLimit(currentLimit.get()); + currentStats.setRejectedConnections(rejectedConnections.get()); + currentStats.setTotalRequests(totalRequests.get()); + currentStats.setRttNoLoad(rttNoLoad); + currentStats.setRttActual(rttActual); + currentStats.setGradient(calculateGradient()); + currentStats.setQueueSize(calculateQueueSize()); + currentStats.setTimestamp(System.currentTimeMillis()); + return currentStats; + } + + /** + * 基于Gradient算法调整限流阈值 + */ + private void adjustLimit() { + try { + // 获取系统资源信息 + SystemResourceMonitor.SystemResourceInfo resourceInfo = resourceMonitor.getCurrentInfo(); + + // 获取配置参数 + int maxConnections = ExternalGlobalConfig.TrafficProtectionOption.maxConnections; + int minThreshold = ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold; + int maxThreshold = ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold; + + // 计算新的阈值 + int newLimit = calculateNewLimit(resourceInfo, maxConnections, minThreshold, maxThreshold); + + // 更新阈值 + int oldLimit = currentLimit.get(); + if (newLimit != oldLimit) { + currentLimit.set(newLimit); + log.info("限流阈值已调整: {} -> {} (Gradient: {:.3f}, RTT比值: {:.3f}/{:.3f})", + oldLimit, newLimit, calculateGradient(), rttNoLoad, rttActual); + } + + // 更新统计信息 + updateStats(resourceInfo); + + } catch (Exception e) { + log.error("调整限流阈值失败", e); + } + } + + /** + * 计算新的限流阈值 + * 基于Gradient算法：newLimit = currentLimit × gradient + queueSize + */ + private int calculateNewLimit( + SystemResourceMonitor.SystemResourceInfo resourceInfo, + int maxConnections, int minThreshold, int maxThreshold) { + + int currentLimit = this.currentLimit.get(); + + // 计算Gradient值 + double gradient = calculateGradient(); + + // 计算队列大小（允许一定的排队） + int queueSize = calculateQueueSize(); + + // 应用Gradient算法 + double newLimitDouble = currentLimit * gradient + queueSize; + int newLimit = (int) Math.round(newLimitDouble); + + // 限制在有效范围内 + newLimit = Math.max(minThreshold, Math.min(maxThreshold, newLimit)); + + // 考虑系统资源限制 + newLimit = applyResourceConstraints(newLimit, resourceInfo); + + return newLimit; + } + + /** + * 计算Gradient值 + * gradient = RTTNoLoad / RTTactual + * 当gradient = 1时，说明当前请求没有排队 + * 当gradient < 1时，说明当前开始排队了，需要降低limit + */ + private double calculateGradient() { + if (rttActual <= 0) return MAX_GRADIENT; + + double gradient = rttNoLoad / rttActual; + + // 限制在有效范围内 + return Math.max(MIN_GRADIENT, Math.min(MAX_GRADIENT, gradient)); + } + + /** + * 计算队列大小 + * 一般设置为当前limit的平方根，对于小的limit增长很快，大的limit变动平稳 + */ + private int calculateQueueSize() { + int currentLimit = this.currentLimit.get(); + return (int) Math.sqrt(currentLimit); + } + + /** + * 应用系统资源约束 + */ + private int applyResourceConstraints(int newLimit, SystemResourceMonitor.SystemResourceInfo resourceInfo) { + double cpuUsage = resourceInfo.getCpuUsage(); + double memoryUsage = resourceInfo.getMemoryUsage(); + + double cpuThreshold = ExternalGlobalConfig.TrafficProtectionOption.cpuThreshold; + double memoryThreshold = ExternalGlobalConfig.TrafficProtectionOption.memoryThreshold; + + // 如果CPU或内存使用率过高，减少limit + if (cpuUsage > cpuThreshold || memoryUsage > memoryThreshold) { + double reductionFactor = 0.8; // 减少20% + newLimit = (int) (newLimit * reductionFactor); + } + + return newLimit; + } + + /** + * 更新RTT值 + * 使用指数移动平均来平滑RTT变化 + */ + private void updateRtt(double rtt) { + long currentTime = System.currentTimeMillis(); + + // 更新实际RTT + if (rttActual == 0) { + rttActual = rtt; + } else { + rttActual = ALPHA * rtt + (1 - ALPHA) * rttActual; + } + + // 更新无负载RTT（取最小值） + if (rttNoLoad == 0 || rtt < rttNoLoad) { + rttNoLoad = rtt; + } + + // 定期更新无负载RTT（防止长期不更新） + if (currentTime - lastUpdateTime > 60000) { // 1分钟 + updateNoLoadRtt(); + lastUpdateTime = currentTime; + } + } + + /** + * 更新无负载RTT + * 使用最近一段时间的最小RTT作为无负载RTT + */ + private void updateNoLoadRtt() { + if (sampleCount.get() < MIN_SAMPLE_SIZE) return; + + // 计算平均RTT + double avgRtt = (double) totalRtt.get() / sampleCount.get() / 1000.0; // 转换回毫秒 + + // 如果当前RTT接近平均RTT，说明系统负载较低，可以更新无负载RTT + if (rttActual < avgRtt * 1.2) { + rttNoLoad = Math.min(rttNoLoad, rttActual); + } + + // 重置统计 + sampleCount.set(0); + totalRtt.set(0); + } + + private void updateStats(SystemResourceMonitor.SystemResourceInfo resourceInfo) { + RateLimitStats currentStats = stats.get(); + currentStats.setCpuUsage(resourceInfo.getCpuUsage()); + currentStats.setMemoryUsage(resourceInfo.getMemoryUsage()); + currentStats.setSystemLoad(resourceInfo.getSystemLoad()); + currentStats.setJvmMemoryUsage(resourceInfo.getJvmMemoryUsage()); + } + + @Getter + @Accessors(chain = true) + public static class RateLimitStats { + private int currentConnections = 0; + private int currentLimit = 0; + private long rejectedConnections = 0; + private long totalRequests = 0; + private double cpuUsage = 0.0; + private double memoryUsage = 0.0; + private double systemLoad = 0.0; + private double jvmMemoryUsage = 0.0; + private double rttNoLoad = 0.0; + private double rttActual = 0.0; + private double gradient = 0.0; + private int queueSize = 0; + private long timestamp = 0L; + + public RateLimitStats setCurrentConnections(int currentConnections) { + this.currentConnections = currentConnections; + return this; + } + + public RateLimitStats setCurrentLimit(int currentLimit) { + this.currentLimit = currentLimit; + return this; + } + + public RateLimitStats setRejectedConnections(long rejectedConnections) { + this.rejectedConnections = rejectedConnections; + return this; + } + + public RateLimitStats setTotalRequests(long totalRequests) { + this.totalRequests = totalRequests; + return this; + } + + public RateLimitStats setCpuUsage(double cpuUsage) { + this.cpuUsage = cpuUsage; + return this; + } + + public RateLimitStats setMemoryUsage(double memoryUsage) { + this.memoryUsage = memoryUsage; + return this; + } + + public RateLimitStats setSystemLoad(double systemLoad) { + this.systemLoad = systemLoad; + return this; + } + + public RateLimitStats setJvmMemoryUsage(double jvmMemoryUsage) { + this.jvmMemoryUsage = jvmMemoryUsage; + return this; + } + + public RateLimitStats setRttNoLoad(double rttNoLoad) { + this.rttNoLoad = rttNoLoad; + return this; + } + + public RateLimitStats setRttActual(double rttActual) { + this.rttActual = rttActual; + return this; + } + + public RateLimitStats setGradient(double gradient) { + this.gradient = gradient; + return this; + } + + public RateLimitStats setQueueSize(int queueSize) { + this.queueSize = queueSize; + return this; + } + + public RateLimitStats setTimestamp(long timestamp) { + this.timestamp = timestamp; + return this; + } + + public double getRejectionRate() { + if (totalRequests == 0) return 0.0; + return ((double) rejectedConnections / totalRequests) * 100.0; + } + + public double getConnectionRate() { + if (currentLimit == 0) return 0.0; + return ((double) currentConnections / currentLimit) * 100.0; + } + + @Override + public String toString() { + return String.format( + "RateLimitStats{currentConnections=%d, currentLimit=%d, " + + "rejectedConnections=%d, totalRequests=%d, rejectionRate=%.2f%%, " + + "connectionRate=%.2f%%, gradient=%.3f, rttNoLoad=%.3f, rttActual=%.3f, " + + "queueSize=%d, cpuUsage=%.2f%%, memoryUsage=%.2f%%, " + + "systemLoad=%.2f, jvmMemoryUsage=%.2f%%, timestamp=%d}", + currentConnections, currentLimit, rejectedConnections, totalRequests, + getRejectionRate(), getConnectionRate(), gradient, rttNoLoad, rttActual, + queueSize, cpuUsage, memoryUsage, systemLoad, jvmMemoryUsage, timestamp + ); + } + } +} diff --git a/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/Gradient2RateLimiter.java b/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/Gradient2RateLimiter.java new file mode 100644 index 000000000..b9d7f1d94 --- /dev/null +++ b/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/Gradient2RateLimiter.java @@ -0,0 +1,217 @@ +package com.iohao.game.external.core.ratelimit; + +import com.iohao.game.external.core.config.ExternalGlobalConfig; +import com.iohao.game.external.core.monitor.SystemResourceMonitor; +import lombok.Getter; +import lombok.experimental.Accessors; +import lombok.extern.slf4j.Slf4j; + +/** + * Gradient2限流算法实现 + * 将noLoadRTT设置为RTT的指数移动平均，会随着RTT的增大而增大，缓解了过度保护的情况 + */ +@Slf4j +public class Gradient2RateLimiter extends AbstractRateLimiter { + + private static final Gradient2RateLimiter INSTANCE = new Gradient2RateLimiter(); + + /** RTT相关参数 */ + private volatile double rttNoLoad = 0.0; // 无负载时的RTT + private volatile double rttActual = 0.0; // 当前实际RTT + private volatile double longRtt = 0.0; // 长期RTT（指数移动平均） + private volatile long lastUpdateTime = 0L; // 上次更新时间 + + /** 自适应参数 */ + private static final double MIN_GRADIENT = 0.5; // 最小梯度值 + private static final double MAX_GRADIENT = 1.0; // 最大梯度值 + private static final double ALPHA = 0.1; // 平滑因子 + private static final double WINDOW_SIZE = 100.0; // 请求窗口大小 + private static final int MIN_SAMPLE_SIZE = 10; // 最小样本数 + + private Gradient2RateLimiter() { + this.rttNoLoad = 1.0; + this.rttActual = 1.0; + this.longRtt = 1.0; + } + + public static Gradient2RateLimiter getInstance() { + return INSTANCE; + } + + @Override + protected String getAlgorithmName() { + return "Gradient2"; + } + + @Override + protected void adjustLimit() { + try { + SystemResourceMonitor.SystemResourceInfo resourceInfo = resourceMonitor.getCurrentInfo(); + + int maxConnections = ExternalGlobalConfig.TrafficProtectionOption.maxConnections; + int minThreshold = ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold; + int maxThreshold = ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold; + + int newLimit = calculateNewLimit(resourceInfo, maxConnections, minThreshold, maxThreshold); + + int oldLimit = currentLimit.get(); + if (newLimit != oldLimit) { + currentLimit.set(newLimit); + log.info("限流阈值已调整: {} -> {} (Gradient: {:.3f}, LongRTT: {:.3f}, RTT: {:.3f})", + oldLimit, newLimit, calculateGradient(), longRtt, rttActual); + } + + } catch (Exception e) { + log.error("调整限流阈值失败", e); + } + } + + private int calculateNewLimit( + SystemResourceMonitor.SystemResourceInfo resourceInfo, + int maxConnections, int minThreshold, int maxThreshold) { + + int currentLimit = this.currentLimit.get(); + double gradient = calculateGradient(); + int queueSize = calculateQueueSize(); + + double newLimitDouble = currentLimit * gradient + queueSize; + int newLimit = (int) Math.round(newLimitDouble); + + newLimit = Math.max(minThreshold, Math.min(maxThreshold, newLimit)); + newLimit = applyResourceConstraints(newLimit, resourceInfo); + + return newLimit; + } + + private double calculateGradient() { + if (rttActual <= 0) return MAX_GRADIENT; + + double gradient = longRtt / rttActual; + return Math.max(MIN_GRADIENT, Math.min(MAX_GRADIENT, gradient)); + } + + private int calculateQueueSize() { + int currentLimit = this.currentLimit.get(); + return (int) Math.sqrt(currentLimit); + } + + @Override + protected void updateRtt(double rtt) { + long currentTime = System.currentTimeMillis(); + + if (rttActual == 0) { + rttActual = rtt; + } else { + rttActual = ALPHA * rtt + (1 - ALPHA) * rttActual; + } + + if (longRtt == 0) { + longRtt = rtt; + } else { + longRtt = (1 - 1.0 / WINDOW_SIZE) * longRtt + (1.0 / WINDOW_SIZE) * rtt; + } + + if (rttNoLoad == 0 || rtt < rttNoLoad) { + rttNoLoad = rtt; + } + + if (currentTime - lastUpdateTime > 60000) { + updateNoLoadRtt(); + lastUpdateTime = currentTime; + } + } + + private void updateNoLoadRtt() { + if (sampleCount.get() < MIN_SAMPLE_SIZE) return; + + double avgRtt = (double) totalRtt.get() / sampleCount.get() / 1000.0; + + if (rttActual < avgRtt * 1.2) { + rttNoLoad = Math.min(rttNoLoad, rttActual); + } + + sampleCount.set(0); + totalRtt.set(0); + } + + @Override + protected void updateStats(BaseRateLimitStats stats) { + SystemResourceMonitor.SystemResourceInfo resourceInfo = resourceMonitor.getCurrentInfo(); + stats.setCpuUsage(resourceInfo.getCpuUsage()); + stats.setMemoryUsage(resourceInfo.getMemoryUsage()); + stats.setSystemLoad(resourceInfo.getSystemLoad()); + stats.setJvmMemoryUsage(resourceInfo.getJvmMemoryUsage()); + + if (stats instanceof Gradient2RateLimitStats) { + Gradient2RateLimitStats gradient2Stats = (Gradient2RateLimitStats) stats; + gradient2Stats.setRttNoLoad(rttNoLoad); + gradient2Stats.setRttActual(rttActual); + gradient2Stats.setLongRtt(longRtt); + gradient2Stats.setGradient(calculateGradient()); + gradient2Stats.setQueueSize(calculateQueueSize()); + } + } + + @Override + public RateLimitStats getStats() { + Gradient2RateLimitStats currentStats = new Gradient2RateLimitStats(); + currentStats.setCurrentConnections(currentConnections.get()); + currentStats.setCurrentLimit(currentLimit.get()); + currentStats.setRejectedConnections(rejectedConnections.get()); + currentStats.setTotalRequests(totalRequests.get()); + currentStats.setTimestamp(System.currentTimeMillis()); + + updateStats(currentStats); + + return currentStats; + } + + @Getter + @Accessors(chain = true) + public static class Gradient2RateLimitStats extends BaseRateLimitStats { + private double rttNoLoad = 0.0; + private double rttActual = 0.0; + private double longRtt = 0.0; + private double gradient = 0.0; + private int queueSize = 0; + + public Gradient2RateLimitStats setRttNoLoad(double rttNoLoad) { + this.rttNoLoad = rttNoLoad; + return this; + } + + public Gradient2RateLimitStats setRttActual(double rttActual) { + this.rttActual = rttActual; + return this; + } + + public Gradient2RateLimitStats setLongRtt(double longRtt) { + this.longRtt = longRtt; + return this; + } + + public Gradient2RateLimitStats setGradient(double gradient) { + this.gradient = gradient; + return this; + } + + public Gradient2RateLimitStats setQueueSize(int queueSize) { + this.queueSize = queueSize; + return this; + } + + @Override + public String toString() { + return String.format( + "Gradient2RateLimitStats{currentConnections=%d, currentLimit=%d, " + + "rejectedConnections=%d, totalRequests=%d, rejectionRate=%.2f%%, " + + "connectionRate=%.2f%%, gradient=%.3f, rttNoLoad=%.3f, rttActual=%.3f, " + + "longRtt=%.3f, queueSize=%d, cpuUsage=%.2f%%, memoryUsage=%.2f%%, " + + "systemLoad=%.2f, jvmMemoryUsage=%.2f%%, timestamp=%d}", + getCurrentConnections(), getCurrentLimit(), getRejectedConnections(), getTotalRequests(), + getRejectionRate(), getConnectionRate(), gradient, rttNoLoad, rttActual, + longRtt, queueSize, getCpuUsage(), getMemoryUsage(), getSystemLoad(), getJvmMemoryUsage(), getTimestamp() + ); + } + } +} diff --git a/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/RateLimiter.java b/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/RateLimiter.java new file mode 100644 index 000000000..76e8d95e9 --- /dev/null +++ b/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/RateLimiter.java @@ -0,0 +1,152 @@ +package com.iohao.game.external.core.ratelimit; + +import com.iohao.game.external.core.monitor.SystemResourceMonitor; + +/** + * 限流器抽象接口 + * 为各种自适应限流算法提供统一的抽象 + * + * 参考：https://gummary.github.io/post/netfix%E8%87%AA%E9%80%82%E5%BA%94%E9%99%90%E6%B5%81%E7%AE%97%E6%B3%95/ + */ +public interface RateLimiter { + + /** + * 启动限流器 + */ + void start(); + + /** + * 停止限流器 + */ + void stop(); + + /** + * 尝试获取连接许可 + * + * @return true 允许连接，false 拒绝连接 + */ + boolean tryAcquire(); + + /** + * 增加连接数 + */ + void incrementConnections(); + + /** + * 减少连接数 + */ + void decrementConnections(); + + /** + * 获取当前连接数 + * + * @return 当前连接数 + */ + int getCurrentConnections(); + + /** + * 获取当前限流阈值 + * + * @return 当前限流阈值 + */ + int getCurrentLimit(); + + /** + * 记录RTT样本 + * + * @param rtt 响应时间（毫秒） + */ + void recordRtt(double rtt); + + /** + * 获取限流统计信息 + * + * @return 限流统计信息 + */ + RateLimitStats getStats(); + + /** + * 限流统计信息接口 + */ + interface RateLimitStats { + /** + * 获取当前连接数 + */ + int getCurrentConnections(); + + /** + * 获取当前限流阈值 + */ + int getCurrentLimit(); + + /** + * 获取被拒绝的连接数 + */ + long getRejectedConnections(); + + /** + * 获取总连接请求数 + */ + long getTotalRequests(); + + /** + * 获取CPU使用率 + */ + double getCpuUsage(); + + /** + * 获取内存使用率 + */ + double getMemoryUsage(); + + /** + * 获取系统负载 + */ + double getSystemLoad(); + + /** + * 获取JVM内存使用率 + */ + double getJvmMemoryUsage(); + + /** + * 获取拒绝率 + */ + double getRejectionRate(); + + /** + * 获取连接率 + */ + double getConnectionRate(); + + /** + * 获取时间戳 + */ + long getTimestamp(); + } + + /** + * 限流器工厂接口 + */ + interface Factory { + /** + * 创建限流器 + * + * @param algorithm 算法类型 + * @return 限流器实例 + */ + RateLimiter createRateLimiter(Algorithm algorithm); + + /** + * 算法类型枚举 + */ + enum Algorithm { + /** Gradient算法 */ + GRADIENT, + /** Gradient2算法 */ + GRADIENT2, + /** Vegas算法 */ + VEGAS + } + } +} diff --git a/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/VegasRateLimiter.java b/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/VegasRateLimiter.java new file mode 100644 index 000000000..a2a41c173 --- /dev/null +++ b/external/external-core/src/main/java/com/iohao/game/external/core/ratelimit/VegasRateLimiter.java @@ -0,0 +1,248 @@ +package com.iohao.game.external.core.ratelimit; + +import com.iohao.game.external.core.config.ExternalGlobalConfig; +import com.iohao.game.external.core.monitor.SystemResourceMonitor; +import lombok.Getter; +import lombok.experimental.Accessors; +import lombok.extern.slf4j.Slf4j; + +/** + * Vegas限流算法实现 + * 启发自TCP的Vegas拥塞避免算法，根据RTT来判断网络是否出现拥塞 + */ +@Slf4j +public class VegasRateLimiter extends AbstractRateLimiter { + + private static final VegasRateLimiter INSTANCE = new VegasRateLimiter(); + + /** RTT相关参数 */ + private volatile double rttNoLoad = 0.0; // 无负载时的RTT + private volatile double rttActual = 0.0; // 当前实际RTT + private volatile long lastUpdateTime = 0L; // 上次更新时间 + + /** Vegas算法参数 */ + private static final double ALPHA = 0.1; // 平滑因子 + private static final int MIN_SAMPLE_SIZE = 10; // 最小样本数 + + private VegasRateLimiter() { + this.rttNoLoad = 1.0; + this.rttActual = 1.0; + } + + public static VegasRateLimiter getInstance() { + return INSTANCE; + } + + @Override + protected String getAlgorithmName() { + return "Vegas"; + } + + @Override + protected void adjustLimit() { + try { + SystemResourceMonitor.SystemResourceInfo resourceInfo = resourceMonitor.getCurrentInfo(); + + int maxConnections = ExternalGlobalConfig.TrafficProtectionOption.maxConnections; + int minThreshold = ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold; + int maxThreshold = ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold; + + int newLimit = calculateNewLimit(resourceInfo, maxConnections, minThreshold, maxThreshold); + + int oldLimit = currentLimit.get(); + if (newLimit != oldLimit) { + currentLimit.set(newLimit); + log.info("限流阈值已调整: {} -> {} (QueueSize: {}, RTT比值: {:.3f}/{:.3f})", + oldLimit, newLimit, calculateQueueSize(), rttNoLoad, rttActual); + } + + } catch (Exception e) { + log.error("调整限流阈值失败", e); + } + } + + /** + * 基于Vegas算法计算新的限流阈值 + * 直接使用等待队列的大小queueSize进行判断是否限流 + */ + private int calculateNewLimit( + SystemResourceMonitor.SystemResourceInfo resourceInfo, + int maxConnections, int minThreshold, int maxThreshold) { + + int currentLimit = this.currentLimit.get(); + int queueSize = calculateQueueSize(); + + // 计算动态调整阈值 + double alpha = 3 * Math.log10(currentLimit); + double beta = 6 * Math.log10(currentLimit); + double threshold = Math.log10(currentLimit); + + int newLimit = currentLimit; + + // Vegas算法核心逻辑 + if (queueSize < threshold) { + // 队列大小小于阈值，扩大窗口 + newLimit = currentLimit + (int) beta; + } else if (queueSize < alpha) { + // 阈值 < 队列大小 < alpha，适度增加 + newLimit = currentLimit + (int) Math.log10(currentLimit); + } else if (queueSize > beta) { + // 队列大小 > beta，缩小窗口 + newLimit = currentLimit - (int) Math.log10(currentLimit); + } + // 其他情况保持当前值 + + // 限制在有效范围内 + newLimit = Math.max(minThreshold, Math.min(maxThreshold, newLimit)); + + // 考虑系统资源限制 + newLimit = applyResourceConstraints(newLimit, resourceInfo); + + return newLimit; + } + + /** + * 计算队列大小 + * queueSize = limit × (1 - RttNoLoad / RTTactual) + */ + private int calculateQueueSize() { + if (rttActual <= 0) return 0; + + int currentLimit = this.currentLimit.get(); + double queueSize = currentLimit * (1 - rttNoLoad / rttActual); + + return Math.max(0, (int) Math.round(queueSize)); + } + + @Override + protected void updateRtt(double rtt) { + long currentTime = System.currentTimeMillis(); + + // 更新实际RTT（使用指数移动平均） + if (rttActual == 0) { + rttActual = rtt; + } else { + rttActual = ALPHA * rtt + (1 - ALPHA) * rttActual; + } + + // 更新无负载RTT（取最小值） + if (rttNoLoad == 0 || rtt < rttNoLoad) { + rttNoLoad = rtt; + } + + // 定期更新无负载RTT（防止长期不更新） + if (currentTime - lastUpdateTime > 60000) { + updateNoLoadRtt(); + lastUpdateTime = currentTime; + } + } + + /** + * 更新无负载RTT + */ + private void updateNoLoadRtt() { + if (sampleCount.get() < MIN_SAMPLE_SIZE) return; + + double avgRtt = (double) totalRtt.get() / sampleCount.get() / 1000.0; + + if (rttActual < avgRtt * 1.2) { + rttNoLoad = Math.min(rttNoLoad, rttActual); + } + + sampleCount.set(0); + totalRtt.set(0); + } + + @Override + protected void updateStats(BaseRateLimitStats stats) { + SystemResourceMonitor.SystemResourceInfo resourceInfo = resourceMonitor.getCurrentInfo(); + stats.setCpuUsage(resourceInfo.getCpuUsage()); + stats.setMemoryUsage(resourceInfo.getMemoryUsage()); + stats.setSystemLoad(resourceInfo.getSystemLoad()); + stats.setJvmMemoryUsage(resourceInfo.getJvmMemoryUsage()); + + if (stats instanceof VegasRateLimitStats) { + VegasRateLimitStats vegasStats = (VegasRateLimitStats) stats; + vegasStats.setRttNoLoad(rttNoLoad); + vegasStats.setRttActual(rttActual); + vegasStats.setQueueSize(calculateQueueSize()); + vegasStats.setAlpha(3 * Math.log10(currentLimit.get())); + vegasStats.setBeta(6 * Math.log10(currentLimit.get())); + vegasStats.setThreshold(Math.log10(currentLimit.get())); + } + } + + @Override + public RateLimitStats getStats() { + VegasRateLimitStats currentStats = new VegasRateLimitStats(); + currentStats.setCurrentConnections(currentConnections.get()); + currentStats.setCurrentLimit(currentLimit.get()); + currentStats.setRejectedConnections(rejectedConnections.get()); + currentStats.setTotalRequests(totalRequests.get()); + currentStats.setTimestamp(System.currentTimeMillis()); + + updateStats(currentStats); + + return currentStats; + } + + /** + * Vegas限流统计信息 + */ + @Getter + @Accessors(chain = true) + public static class VegasRateLimitStats extends BaseRateLimitStats { + private double rttNoLoad = 0.0; + private double rttActual = 0.0; + private int queueSize = 0; + private double alpha = 0.0; + private double beta = 0.0; + private double threshold = 0.0; + + public VegasRateLimitStats setRttNoLoad(double rttNoLoad) { + this.rttNoLoad = rttNoLoad; + return this; + } + + public VegasRateLimitStats setRttActual(double rttActual) { + this.rttActual = rttActual; + return this; + } + + public VegasRateLimitStats setQueueSize(int queueSize) { + this.queueSize = queueSize; + return this; + } + + public VegasRateLimitStats setAlpha(double alpha) { + this.alpha = alpha; + return this; + } + + public VegasRateLimitStats setBeta(double beta) { + this.beta = beta; + return this; + } + + public VegasRateLimitStats setThreshold(double threshold) { + this.threshold = threshold; + return this; + } + + @Override + public String toString() { + return String.format( + "VegasRateLimitStats{currentConnections=%d, currentLimit=%d, " + + "rejectedConnections=%d, totalRequests=%d, rejectionRate=%.2f%%, " + + "connectionRate=%.2f%%, queueSize=%d, alpha=%.3f, beta=%.3f, " + + "threshold=%.3f, rttNoLoad=%.3f, rttActual=%.3f, " + + "cpuUsage=%.2f%%, memoryUsage=%.2f%%, systemLoad=%.2f, " + + "jvmMemoryUsage=%.2f%%, timestamp=%d}", + getCurrentConnections(), getCurrentLimit(), getRejectedConnections(), getTotalRequests(), + getRejectionRate(), getConnectionRate(), queueSize, alpha, beta, threshold, + rttNoLoad, rttActual, getCpuUsage(), getMemoryUsage(), getSystemLoad(), + getJvmMemoryUsage(), getTimestamp() + ); + } + } +} diff --git a/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/Gradient2RateLimiterTest.java b/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/Gradient2RateLimiterTest.java new file mode 100644 index 000000000..58eabe5d6 --- /dev/null +++ b/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/Gradient2RateLimiterTest.java @@ -0,0 +1,246 @@ +package com.iohao.game.external.core.ratelimit; + +import com.iohao.game.external.core.config.ExternalGlobalConfig; +import com.iohao.game.external.core.monitor.SystemResourceMonitor; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +/** + * Gradient2限流算法单元测试 + */ +@ExtendWith(MockitoExtension.class) +class Gradient2RateLimiterTest { + + @Mock + private SystemResourceMonitor mockResourceMonitor; + + @Mock + private SystemResourceMonitor.SystemResourceInfo mockResourceInfo; + + private Gradient2RateLimiter rateLimiter; + + @BeforeEach + void setUp() { + // 设置配置参数 + ExternalGlobalConfig.TrafficProtectionOption.enableAdaptiveRateLimit = true; + ExternalGlobalConfig.TrafficProtectionOption.maxConnections = 1000; + ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold = 100; + ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold = 2000; + ExternalGlobalConfig.TrafficProtectionOption.monitorInterval = 100; + ExternalGlobalConfig.TrafficProtectionOption.cpuThreshold = 80.0; + ExternalGlobalConfig.TrafficProtectionOption.memoryThreshold = 85.0; + + // 创建限流器实例 + rateLimiter = Gradient2RateLimiter.getInstance(); + } + + @Test + void testInitialState() { + // 测试初始状态 + assertEquals(1000, rateLimiter.getCurrentLimit()); + assertEquals(0, rateLimiter.getCurrentConnections()); + assertFalse(rateLimiter.getStats().getRejectionRate() > 0); + } + + @Test + void testTryAcquire() { + // 测试连接获取 + assertTrue(rateLimiter.tryAcquire()); + assertEquals(1, rateLimiter.getCurrentConnections()); + + // 测试连接数达到限制时 + for (int i = 1; i < 1000; i++) { + rateLimiter.incrementConnections(); + } + + assertFalse(rateLimiter.tryAcquire()); + assertEquals(1000, rateLimiter.getCurrentConnections()); + } + + @Test + void testConnectionLifecycle() { + // 测试连接生命周期 + assertTrue(rateLimiter.tryAcquire()); + rateLimiter.incrementConnections(); + assertEquals(1, rateLimiter.getCurrentConnections()); + + rateLimiter.decrementConnections(); + assertEquals(0, rateLimiter.getCurrentConnections()); + } + + @Test + void testRttRecording() { + // 测试RTT记录 + rateLimiter.recordRtt(10.0); + rateLimiter.recordRtt(15.0); + rateLimiter.recordRtt(20.0); + + // 验证RTT统计 + assertTrue(rateLimiter.getStats().getTotalRequests() > 0); + } + + @Test + void testStartAndStop() { + // 测试启动和停止 + assertDoesNotThrow(() -> rateLimiter.start()); + assertDoesNotThrow(() -> rateLimiter.stop()); + } + + @Test + void testStatsCalculation() { + // 测试统计信息计算 + rateLimiter.tryAcquire(); + rateLimiter.incrementConnections(); + rateLimiter.recordRtt(10.0); + + var stats = rateLimiter.getStats(); + assertEquals(1, stats.getCurrentConnections()); + assertEquals(1000, stats.getCurrentLimit()); + assertEquals(1, stats.getTotalRequests()); + assertEquals(0.0, stats.getRejectionRate()); + assertEquals(0.1, stats.getConnectionRate()); + } + + @Test + void testGradient2Calculation() { + // 测试Gradient2值计算 + rateLimiter.recordRtt(1.0); // 无负载RTT + rateLimiter.recordRtt(2.0); // 实际RTT + + // 验证Gradient值在合理范围内 + var stats = rateLimiter.getStats(); + assertTrue(stats.getGradient() >= 0.5 && stats.getGradient() <= 1.0); + } + + @Test + void testLongRttCalculation() { + // 测试长期RTT计算（指数移动平均） + rateLimiter.recordRtt(10.0); + rateLimiter.recordRtt(15.0); + rateLimiter.recordRtt(20.0); + + // 验证长期RTT被正确计算 + var stats = rateLimiter.getStats(); + assertTrue(stats.getLongRtt() > 0); + } + + @Test + void testQueueSizeCalculation() { + // 测试队列大小计算 + int queueSize = (int) Math.sqrt(1000); + assertEquals(31, queueSize); // sqrt(1000) ≈ 31.62 + } + + @Test + void testMultipleConnections() { + // 测试多个连接 + for (int i = 0; i < 100; i++) { + assertTrue(rateLimiter.tryAcquire()); + rateLimiter.incrementConnections(); + } + + assertEquals(100, rateLimiter.getCurrentConnections()); + assertEquals(100, rateLimiter.getStats().getTotalRequests()); + } + + @Test + void testRejectionTracking() { + // 测试拒绝连接跟踪 + // 先填满连接数 + for (int i = 0; i < 1000; i++) { + rateLimiter.incrementConnections(); + } + + // 尝试获取新连接，应该被拒绝 + assertFalse(rateLimiter.tryAcquire()); + + // 验证拒绝统计 + assertEquals(1, rateLimiter.getStats().getRejectedConnections()); + assertTrue(rateLimiter.getStats().getRejectionRate() > 0); + } + + @Test + void testBoundaryConditions() { + // 测试边界条件 + // 设置最小阈值 + ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold = 500; + + // 验证限流阈值不会低于最小值 + assertTrue(rateLimiter.getCurrentLimit() >= 500); + + // 设置最大阈值 + ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold = 1500; + + // 验证限流阈值不会超过最大值 + assertTrue(rateLimiter.getCurrentLimit() <= 1500); + } + + @Test + void testGradient2SpecificFeatures() { + // 测试Gradient2特有的功能 + rateLimiter.recordRtt(5.0); + rateLimiter.recordRtt(10.0); + rateLimiter.recordRtt(15.0); + + var stats = rateLimiter.getStats(); + + // 验证Gradient2特有的字段 + assertTrue(stats.getLongRtt() > 0); + assertTrue(stats.getGradient() >= 0.5 && stats.getGradient() <= 1.0); + assertTrue(stats.getQueueSize() > 0); + } + + @Test + void testExponentialMovingAverage() { + // 测试指数移动平均计算 + // 连续记录多个RTT值 + for (int i = 1; i <= 10; i++) { + rateLimiter.recordRtt(i * 10.0); + } + + var stats = rateLimiter.getStats(); + + // 验证长期RTT被正确计算 + assertTrue(stats.getLongRtt() > 0); + assertTrue(stats.getLongRtt() <= 100.0); // 最大值应该是100 + } + + @Test + void testNoLoadRttUpdate() { + // 测试无负载RTT更新 + rateLimiter.recordRtt(1.0); // 设置一个很小的RTT + + // 等待一段时间后再次记录RTT + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + rateLimiter.recordRtt(2.0); + + var stats = rateLimiter.getStats(); + assertTrue(stats.getRttNoLoad() > 0); + } + + @Test + void testGradient2StatsExtension() { + // 测试Gradient2统计信息扩展 + rateLimiter.recordRtt(10.0); + + var stats = rateLimiter.getStats(); + + // 验证Gradient2特有的统计字段 + if (stats instanceof Gradient2RateLimiter.Gradient2RateLimitStats) { + var gradient2Stats = (Gradient2RateLimiter.Gradient2RateLimitStats) stats; + assertTrue(gradient2Stats.getLongRtt() > 0); + assertTrue(gradient2Stats.getGradient() >= 0.5 && gradient2Stats.getGradient() <= 1.0); + } + } +} diff --git a/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/GradientRateLimiterTest.java b/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/GradientRateLimiterTest.java new file mode 100644 index 000000000..a98e74c4b --- /dev/null +++ b/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/GradientRateLimiterTest.java @@ -0,0 +1,182 @@ +package com.iohao.game.external.core.ratelimit; + +import com.iohao.game.external.core.config.ExternalGlobalConfig; +import com.iohao.game.external.core.monitor.SystemResourceMonitor; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +/** + * Gradient限流算法单元测试 + */ +@ExtendWith(MockitoExtension.class) +class GradientRateLimiterTest { + + @Mock + private SystemResourceMonitor mockResourceMonitor; + + @Mock + private SystemResourceMonitor.SystemResourceInfo mockResourceInfo; + + private GradientRateLimiter rateLimiter; + + @BeforeEach + void setUp() { + // 设置配置参数 + ExternalGlobalConfig.TrafficProtectionOption.enableAdaptiveRateLimit = true; + ExternalGlobalConfig.TrafficProtectionOption.maxConnections = 1000; + ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold = 100; + ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold = 2000; + ExternalGlobalConfig.TrafficProtectionOption.monitorInterval = 100; + ExternalGlobalConfig.TrafficProtectionOption.cpuThreshold = 80.0; + ExternalGlobalConfig.TrafficProtectionOption.memoryThreshold = 85.0; + + // 创建限流器实例 + rateLimiter = GradientRateLimiter.getInstance(); + } + + @Test + void testInitialState() { + // 测试初始状态 + assertEquals(1000, rateLimiter.getCurrentLimit()); + assertEquals(0, rateLimiter.getCurrentConnections()); + assertFalse(rateLimiter.getStats().getRejectionRate() > 0); + } + + @Test + void testTryAcquire() { + // 测试连接获取 + assertTrue(rateLimiter.tryAcquire()); + assertEquals(1, rateLimiter.getCurrentConnections()); + + // 测试连接数达到限制时 + for (int i = 1; i < 1000; i++) { + rateLimiter.incrementConnections(); + } + + assertFalse(rateLimiter.tryAcquire()); + assertEquals(1000, rateLimiter.getCurrentConnections()); + } + + @Test + void testConnectionLifecycle() { + // 测试连接生命周期 + assertTrue(rateLimiter.tryAcquire()); + rateLimiter.incrementConnections(); + assertEquals(1, rateLimiter.getCurrentConnections()); + + rateLimiter.decrementConnections(); + assertEquals(0, rateLimiter.getCurrentConnections()); + } + + @Test + void testRttRecording() { + // 测试RTT记录 + rateLimiter.recordRtt(10.0); + rateLimiter.recordRtt(15.0); + rateLimiter.recordRtt(20.0); + + // 验证RTT统计 + assertTrue(rateLimiter.getStats().getTotalRequests() > 0); + } + + @Test + void testStartAndStop() { + // 测试启动和停止 + assertDoesNotThrow(() -> rateLimiter.start()); + assertDoesNotThrow(() -> rateLimiter.stop()); + } + + @Test + void testStatsCalculation() { + // 测试统计信息计算 + rateLimiter.tryAcquire(); + rateLimiter.incrementConnections(); + rateLimiter.recordRtt(10.0); + + var stats = rateLimiter.getStats(); + assertEquals(1, stats.getCurrentConnections()); + assertEquals(1000, stats.getCurrentLimit()); + assertEquals(1, stats.getTotalRequests()); + assertEquals(0.0, stats.getRejectionRate()); + assertEquals(0.1, stats.getConnectionRate()); + } + + @Test + void testResourceConstraints() { + // 测试资源约束 + when(mockResourceInfo.getCpuUsage()).thenReturn(90.0); + when(mockResourceInfo.getMemoryUsage()).thenReturn(90.0); + + // 模拟资源监控器 + // 这里需要注入mock，但由于是单例模式，我们直接测试算法逻辑 + assertTrue(rateLimiter.getCurrentLimit() > 0); + } + + @Test + void testGradientCalculation() { + // 测试Gradient值计算 + rateLimiter.recordRtt(1.0); // 无负载RTT + rateLimiter.recordRtt(2.0); // 实际RTT + + // 验证Gradient值在合理范围内 + var stats = rateLimiter.getStats(); + assertTrue(stats.getGradient() >= 0.5 && stats.getGradient() <= 1.0); + } + + @Test + void testQueueSizeCalculation() { + // 测试队列大小计算 + int queueSize = (int) Math.sqrt(1000); + assertEquals(31, queueSize); // sqrt(1000) ≈ 31.62 + } + + @Test + void testMultipleConnections() { + // 测试多个连接 + for (int i = 0; i < 100; i++) { + assertTrue(rateLimiter.tryAcquire()); + rateLimiter.incrementConnections(); + } + + assertEquals(100, rateLimiter.getCurrentConnections()); + assertEquals(100, rateLimiter.getStats().getTotalRequests()); + } + + @Test + void testRejectionTracking() { + // 测试拒绝连接跟踪 + // 先填满连接数 + for (int i = 0; i < 1000; i++) { + rateLimiter.incrementConnections(); + } + + // 尝试获取新连接，应该被拒绝 + assertFalse(rateLimiter.tryAcquire()); + + // 验证拒绝统计 + assertEquals(1, rateLimiter.getStats().getRejectedConnections()); + assertTrue(rateLimiter.getStats().getRejectionRate() > 0); + } + + @Test + void testBoundaryConditions() { + // 测试边界条件 + // 设置最小阈值 + ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold = 500; + + // 验证限流阈值不会低于最小值 + assertTrue(rateLimiter.getCurrentLimit() >= 500); + + // 设置最大阈值 + ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold = 1500; + + // 验证限流阈值不会超过最大值 + assertTrue(rateLimiter.getCurrentLimit() <= 1500); + } +} diff --git a/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/RateLimiterIntegrationTest.java b/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/RateLimiterIntegrationTest.java new file mode 100644 index 000000000..e24ac21dc --- /dev/null +++ b/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/RateLimiterIntegrationTest.java @@ -0,0 +1,285 @@ +package com.iohao.game.external.core.ratelimit; + +import com.iohao.game.external.core.config.ExternalGlobalConfig; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.junit.jupiter.MockitoExtension; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * 限流器集成测试 + * 测试所有三种算法的基本功能和性能 + */ +@ExtendWith(MockitoExtension.class) +class RateLimiterIntegrationTest { + + private GradientRateLimiter gradientRateLimiter; + private Gradient2RateLimiter gradient2RateLimiter; + private VegasRateLimiter vegasRateLimiter; + + @BeforeEach + void setUp() { + // 设置配置参数 + ExternalGlobalConfig.TrafficProtectionOption.enableAdaptiveRateLimit = true; + ExternalGlobalConfig.TrafficProtectionOption.maxConnections = 1000; + ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold = 100; + ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold = 2000; + ExternalGlobalConfig.TrafficProtectionOption.monitorInterval = 100; + ExternalGlobalConfig.TrafficProtectionOption.cpuThreshold = 80.0; + ExternalGlobalConfig.TrafficProtectionOption.memoryThreshold = 85.0; + + // 创建所有限流器实例 + gradientRateLimiter = GradientRateLimiter.getInstance(); + gradient2RateLimiter = Gradient2RateLimiter.getInstance(); + vegasRateLimiter = VegasRateLimiter.getInstance(); + } + + @Test + void testAllAlgorithmsInitialState() { + // 测试所有算法的初始状态 + assertEquals(1000, gradientRateLimiter.getCurrentLimit()); + assertEquals(1000, gradient2RateLimiter.getCurrentLimit()); + assertEquals(1000, vegasRateLimiter.getCurrentLimit()); + + assertEquals(0, gradientRateLimiter.getCurrentConnections()); + assertEquals(0, gradient2RateLimiter.getCurrentConnections()); + assertEquals(0, vegasRateLimiter.getCurrentConnections()); + } + + @Test + void testAllAlgorithmsConnectionLifecycle() { + // 测试所有算法的连接生命周期 + // Gradient算法 + assertTrue(gradientRateLimiter.tryAcquire()); + gradientRateLimiter.incrementConnections(); + assertEquals(1, gradientRateLimiter.getCurrentConnections()); + gradientRateLimiter.decrementConnections(); + assertEquals(0, gradientRateLimiter.getCurrentConnections()); + + // Gradient2算法 + assertTrue(gradient2RateLimiter.tryAcquire()); + gradient2RateLimiter.incrementConnections(); + assertEquals(1, gradient2RateLimiter.getCurrentConnections()); + gradient2RateLimiter.decrementConnections(); + assertEquals(0, gradient2RateLimiter.getCurrentConnections()); + + // Vegas算法 + assertTrue(vegasRateLimiter.tryAcquire()); + vegasRateLimiter.incrementConnections(); + assertEquals(1, vegasRateLimiter.getCurrentConnections()); + vegasRateLimiter.decrementConnections(); + assertEquals(0, vegasRateLimiter.getCurrentConnections()); + } + + @Test + void testAllAlgorithmsRttRecording() { + // 测试所有算法的RTT记录 + double[] rttValues = {10.0, 15.0, 20.0, 25.0, 30.0}; + + for (double rtt : rttValues) { + gradientRateLimiter.recordRtt(rtt); + gradient2RateLimiter.recordRtt(rtt); + vegasRateLimiter.recordRtt(rtt); + } + + // 验证所有算法都正确记录了RTT + assertTrue(gradientRateLimiter.getStats().getTotalRequests() > 0); + assertTrue(gradient2RateLimiter.getStats().getTotalRequests() > 0); + assertTrue(vegasRateLimiter.getStats().getTotalRequests() > 0); + } + + @Test + void testAllAlgorithmsStartAndStop() { + // 测试所有算法的启动和停止 + assertDoesNotThrow(() -> { + gradientRateLimiter.start(); + gradient2RateLimiter.start(); + vegasRateLimiter.start(); + }); + + assertDoesNotThrow(() -> { + gradientRateLimiter.stop(); + gradient2RateLimiter.stop(); + vegasRateLimiter.stop(); + }); + } + + @Test + void testAllAlgorithmsStatsCalculation() { + // 测试所有算法的统计信息计算 + // 添加一些连接和RTT数据 + for (int i = 0; i < 10; i++) { + gradientRateLimiter.tryAcquire(); + gradientRateLimiter.incrementConnections(); + gradientRateLimiter.recordRtt(10.0 + i); + + gradient2RateLimiter.tryAcquire(); + gradient2RateLimiter.incrementConnections(); + gradient2RateLimiter.recordRtt(10.0 + i); + + vegasRateLimiter.tryAcquire(); + vegasRateLimiter.incrementConnections(); + vegasRateLimiter.recordRtt(10.0 + i); + } + + // 验证统计信息 + var gradientStats = gradientRateLimiter.getStats(); + var gradient2Stats = gradient2RateLimiter.getStats(); + var vegasStats = vegasRateLimiter.getStats(); + + assertEquals(10, gradientStats.getCurrentConnections()); + assertEquals(10, gradient2Stats.getCurrentConnections()); + assertEquals(10, vegasStats.getCurrentConnections()); + + assertEquals(10, gradientStats.getTotalRequests()); + assertEquals(10, gradient2Stats.getTotalRequests()); + assertEquals(10, vegasStats.getTotalRequests()); + } + + @Test + void testAllAlgorithmsRejectionHandling() { + // 测试所有算法的拒绝连接处理 + // 先填满连接数 + for (int i = 0; i < 1000; i++) { + gradientRateLimiter.incrementConnections(); + gradient2RateLimiter.incrementConnections(); + vegasRateLimiter.incrementConnections(); + } + + // 尝试获取新连接，应该被拒绝 + assertFalse(gradientRateLimiter.tryAcquire()); + assertFalse(gradient2RateLimiter.tryAcquire()); + assertFalse(vegasRateLimiter.tryAcquire()); + + // 验证拒绝统计 + assertEquals(1, gradientRateLimiter.getStats().getRejectedConnections()); + assertEquals(1, gradient2RateLimiter.getStats().getRejectedConnections()); + assertEquals(1, vegasRateLimiter.getStats().getRejectedConnections()); + } + + @Test + void testAllAlgorithmsBoundaryConditions() { + // 测试所有算法的边界条件 + // 设置最小和最大阈值 + ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold = 500; + ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold = 1500; + + // 验证所有算法的限流阈值都在有效范围内 + assertTrue(gradientRateLimiter.getCurrentLimit() >= 500); + assertTrue(gradient2RateLimiter.getCurrentLimit() >= 500); + assertTrue(vegasRateLimiter.getCurrentLimit() >= 500); + + assertTrue(gradientRateLimiter.getCurrentLimit() <= 1500); + assertTrue(gradient2RateLimiter.getCurrentLimit() <= 1500); + assertTrue(vegasRateLimiter.getCurrentLimit() <= 1500); + } + + @Test + void testAllAlgorithmsPerformance() { + // 测试所有算法的性能 + long startTime = System.currentTimeMillis(); + + // 执行大量操作 + for (int i = 0; i < 1000; i++) { + gradientRateLimiter.tryAcquire(); + gradient2RateLimiter.tryAcquire(); + vegasRateLimiter.tryAcquire(); + + if (i % 100 == 0) { + gradientRateLimiter.recordRtt(i * 0.1); + gradient2RateLimiter.recordRtt(i * 0.1); + vegasRateLimiter.recordRtt(i * 0.1); + } + } + + long endTime = System.currentTimeMillis(); + long duration = endTime - startTime; + + // 验证性能（应该在合理时间内完成） + assertTrue(duration < 5000, "性能测试应该在5秒内完成，实际耗时: " + duration + "ms"); + } + + @Test + void testAllAlgorithmsConcurrentAccess() { + // 测试所有算法的并发访问 + int threadCount = 10; + int operationsPerThread = 100; + + Thread[] threads = new Thread[threadCount]; + + for (int i = 0; i < threadCount; i++) { + final int threadId = i; + threads[i] = new Thread(() -> { + for (int j = 0; j < operationsPerThread; j++) { + try { + gradientRateLimiter.tryAcquire(); + gradient2RateLimiter.tryAcquire(); + vegasRateLimiter.tryAcquire(); + + if (j % 10 == 0) { + gradientRateLimiter.recordRtt(threadId * 10.0 + j); + gradient2RateLimiter.recordRtt(threadId * 10.0 + j); + vegasRateLimiter.recordRtt(threadId * 10.0 + j); + } + } catch (Exception e) { + // 忽略异常，继续测试 + } + } + }); + } + + // 启动所有线程 + for (Thread thread : threads) { + thread.start(); + } + + // 等待所有线程完成 + for (Thread thread : threads) { + try { + thread.join(5000); // 最多等待5秒 + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + // 验证所有算法都能正确处理并发访问 + assertTrue(gradientRateLimiter.getStats().getTotalRequests() > 0); + assertTrue(gradient2RateLimiter.getStats().getTotalRequests() > 0); + assertTrue(vegasRateLimiter.getStats().getTotalRequests() > 0); + } + + @Test + void testAllAlgorithmsMemoryUsage() { + // 测试所有算法的内存使用 + Runtime runtime = Runtime.getRuntime(); + + // 记录初始内存 + long initialMemory = runtime.totalMemory() - runtime.freeMemory(); + + // 执行大量操作 + for (int i = 0; i < 10000; i++) { + gradientRateLimiter.tryAcquire(); + gradient2RateLimiter.tryAcquire(); + vegasRateLimiter.tryAcquire(); + + if (i % 100 == 0) { + gradientRateLimiter.recordRtt(i * 0.01); + gradient2RateLimiter.recordRtt(i * 0.01); + vegasRateLimiter.recordRtt(i * 0.01); + } + } + + // 强制垃圾回收 + System.gc(); + + // 记录最终内存 + long finalMemory = runtime.totalMemory() - runtime.freeMemory(); + long memoryIncrease = finalMemory - initialMemory; + + // 验证内存使用在合理范围内（增加不超过100MB） + assertTrue(memoryIncrease < 100 * 1024 * 1024, + "内存使用增加过多: " + (memoryIncrease / 1024 / 1024) + "MB"); + } +} diff --git a/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/VegasRateLimiterTest.java b/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/VegasRateLimiterTest.java new file mode 100644 index 000000000..ce8cc6a66 --- /dev/null +++ b/external/external-core/src/test/java/com/iohao/game/external/core/ratelimit/VegasRateLimiterTest.java @@ -0,0 +1,291 @@ +package com.iohao.game.external.core.ratelimit; + +import com.iohao.game.external.core.config.ExternalGlobalConfig; +import com.iohao.game.external.core.monitor.SystemResourceMonitor; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +/** + * Vegas限流算法单元测试 + */ +@ExtendWith(MockitoExtension.class) +class VegasRateLimiterTest { + + @Mock + private SystemResourceMonitor mockResourceMonitor; + + @Mock + private SystemResourceMonitor.SystemResourceInfo mockResourceInfo; + + private VegasRateLimiter rateLimiter; + + @BeforeEach + void setUp() { + // 设置配置参数 + ExternalGlobalConfig.TrafficProtectionOption.enableAdaptiveRateLimit = true; + ExternalGlobalConfig.TrafficProtectionOption.maxConnections = 1000; + ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold = 100; + ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold = 2000; + ExternalGlobalConfig.TrafficProtectionOption.monitorInterval = 100; + ExternalGlobalConfig.TrafficProtectionOption.cpuThreshold = 80.0; + ExternalGlobalConfig.TrafficProtectionOption.memoryThreshold = 85.0; + + // 创建限流器实例 + rateLimiter = VegasRateLimiter.getInstance(); + } + + @Test + void testInitialState() { + // 测试初始状态 + assertEquals(1000, rateLimiter.getCurrentLimit()); + assertEquals(0, rateLimiter.getCurrentConnections()); + assertFalse(rateLimiter.getStats().getRejectionRate() > 0); + } + + @Test + void testTryAcquire() { + // 测试连接获取 + assertTrue(rateLimiter.tryAcquire()); + assertEquals(1, rateLimiter.getCurrentConnections()); + + // 测试连接数达到限制时 + for (int i = 1; i < 1000; i++) { + rateLimiter.incrementConnections(); + } + + assertFalse(rateLimiter.tryAcquire()); + assertEquals(1000, rateLimiter.getCurrentConnections()); + } + + @Test + void testConnectionLifecycle() { + // 测试连接生命周期 + assertTrue(rateLimiter.tryAcquire()); + rateLimiter.incrementConnections(); + assertEquals(1, rateLimiter.getCurrentConnections()); + + rateLimiter.decrementConnections(); + assertEquals(0, rateLimiter.getCurrentConnections()); + } + + @Test + void testRttRecording() { + // 测试RTT记录 + rateLimiter.recordRtt(10.0); + rateLimiter.recordRtt(15.0); + rateLimiter.recordRtt(20.0); + + // 验证RTT统计 + assertTrue(rateLimiter.getStats().getTotalRequests() > 0); + } + + @Test + void testStartAndStop() { + // 测试启动和停止 + assertDoesNotThrow(() -> rateLimiter.start()); + assertDoesNotThrow(() -> rateLimiter.stop()); + } + + @Test + void testStatsCalculation() { + // 测试统计信息计算 + rateLimiter.tryAcquire(); + rateLimiter.incrementConnections(); + rateLimiter.recordRtt(10.0); + + var stats = rateLimiter.getStats(); + assertEquals(1, stats.getCurrentConnections()); + assertEquals(1000, stats.getCurrentLimit()); + assertEquals(1, stats.getTotalRequests()); + assertEquals(0.0, stats.getRejectionRate()); + assertEquals(0.1, stats.getConnectionRate()); + } + + @Test + void testVegasAlgorithm() { + // 测试Vegas算法核心逻辑 + rateLimiter.recordRtt(1.0); // 无负载RTT + rateLimiter.recordRtt(2.0); // 实际RTT + + // 验证Vegas算法特有的统计字段 + var stats = rateLimiter.getStats(); + assertTrue(stats.getQueueSize() >= 0); + } + + @Test + void testQueueSizeCalculation() { + // 测试队列大小计算 + // queueSize = limit × (1 - RttNoLoad / RTTactual) + rateLimiter.recordRtt(1.0); // 无负载RTT + rateLimiter.recordRtt(2.0); // 实际RTT + + var stats = rateLimiter.getStats(); + int queueSize = stats.getQueueSize(); + + // 当RTTactual > RttNoLoad时，queueSize应该 > 0 + assertTrue(queueSize >= 0); + } + + @Test + void testMultipleConnections() { + // 测试多个连接 + for (int i = 0; i < 100; i++) { + assertTrue(rateLimiter.tryAcquire()); + rateLimiter.incrementConnections(); + } + + assertEquals(100, rateLimiter.getCurrentConnections()); + assertEquals(100, rateLimiter.getStats().getTotalRequests()); + } + + @Test + void testRejectionTracking() { + // 测试拒绝连接跟踪 + // 先填满连接数 + for (int i = 0; i < 1000; i++) { + rateLimiter.incrementConnections(); + } + + // 尝试获取新连接，应该被拒绝 + assertFalse(rateLimiter.tryAcquire()); + + // 验证拒绝统计 + assertEquals(1, rateLimiter.getStats().getRejectedConnections()); + assertTrue(rateLimiter.getStats().getRejectionRate() > 0); + } + + @Test + void testBoundaryConditions() { + // 测试边界条件 + // 设置最小阈值 + ExternalGlobalConfig.TrafficProtectionOption.minRateLimitThreshold = 500; + + // 验证限流阈值不会低于最小值 + assertTrue(rateLimiter.getCurrentLimit() >= 500); + + // 设置最大阈值 + ExternalGlobalConfig.TrafficProtectionOption.maxRateLimitThreshold = 1500; + + // 验证限流阈值不会超过最大值 + assertTrue(rateLimiter.getCurrentLimit() <= 1500); + } + + @Test + void testVegasSpecificFeatures() { + // 测试Vegas特有的功能 + rateLimiter.recordRtt(5.0); + rateLimiter.recordRtt(10.0); + rateLimiter.recordRtt(15.0); + + var stats = rateLimiter.getStats(); + + // 验证Vegas特有的字段 + assertTrue(stats.getQueueSize() >= 0); + } + + @Test + void testDynamicThresholds() { + // 测试动态阈值计算 + // alpha = 3 * log10(limit) + // beta = 6 * log10(limit) + // threshold = log10(limit) + + int currentLimit = 1000; + double expectedAlpha = 3 * Math.log10(currentLimit); + double expectedBeta = 6 * Math.log10(currentLimit); + double expectedThreshold = Math.log10(currentLimit); + + // 验证阈值计算 + assertEquals(expectedAlpha, 3 * Math.log10(1000), 0.001); + assertEquals(expectedBeta, 6 * Math.log10(1000), 0.001); + assertEquals(expectedThreshold, Math.log10(1000), 0.001); + } + + @Test + void testVegasAlgorithmLogic() { + // 测试Vegas算法的核心逻辑 + // 这里我们测试算法的数学计算部分 + + int currentLimit = 1000; + double alpha = 3 * Math.log10(currentLimit); + double beta = 6 * Math.log10(currentLimit); + double threshold = Math.log10(currentLimit); + + // 验证阈值关系 + assertTrue(alpha > threshold); + assertTrue(beta > alpha); + assertTrue(threshold > 0); + } + + @Test + void testQueueSizeBoundary() { + // 测试队列大小边界条件 + rateLimiter.recordRtt(1.0); // 无负载RTT + + // 当RTTactual = RttNoLoad时，queueSize应该 = 0 + rateLimiter.recordRtt(1.0); + + var stats = rateLimiter.getStats(); + int queueSize = stats.getQueueSize(); + + // 由于浮点数精度问题，我们检查是否接近0 + assertTrue(queueSize >= 0); + } + + @Test + void testVegasStatsExtension() { + // 测试Vegas统计信息扩展 + rateLimiter.recordRtt(10.0); + + var stats = rateLimiter.getStats(); + + // 验证Vegas特有的统计字段 + if (stats instanceof VegasRateLimiter.VegasRateLimitStats) { + var vegasStats = (VegasRateLimiter.VegasRateLimitStats) stats; + assertTrue(vegasStats.getQueueSize() >= 0); + assertTrue(vegasStats.getAlpha() > 0); + assertTrue(vegasStats.getBeta() > 0); + assertTrue(vegasStats.getThreshold() > 0); + } + } + + @Test + void testNoLoadRttUpdate() { + // 测试无负载RTT更新 + rateLimiter.recordRtt(1.0); // 设置一个很小的RTT + + // 等待一段时间后再次记录RTT + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + rateLimiter.recordRtt(2.0); + + var stats = rateLimiter.getStats(); + assertTrue(stats.getRttNoLoad() > 0); + } + + @Test + void testVegasAlgorithmConvergence() { + // 测试Vegas算法的收敛性 + // 连续记录多个RTT值，观察算法的收敛行为 + + for (int i = 1; i <= 20; i++) { + rateLimiter.recordRtt(i * 5.0); + } + + var stats = rateLimiter.getStats(); + + // 验证算法能够收敛到合理的值 + assertTrue(stats.getQueueSize() >= 0); + assertTrue(stats.getCurrentLimit() > 0); + } +} diff --git a/external/external-core/src/test/resources/test-config.properties b/external/external-core/src/test/resources/test-config.properties new file mode 100644 index 000000000..d91e962e4 --- /dev/null +++ b/external/external-core/src/test/resources/test-config.properties @@ -0,0 +1,24 @@ +# 测试配置文件 +# 用于限流器单元测试的配置参数 + +# 流量保护配置 +traffic.protection.enable=true +traffic.protection.max.connections=1000 +traffic.protection.min.threshold=100 +traffic.protection.max.threshold=2000 +traffic.protection.monitor.interval=100 + +# 自适应限流配置 +adaptive.rate.limit.enable=true +adaptive.rate.limit.window.size=60 +adaptive.rate.limit.alpha=0.1 + +# 系统资源阈值 +system.cpu.threshold=80.0 +system.memory.threshold=85.0 + +# 测试参数 +test.thread.count=10 +test.operations.per.thread=100 +test.performance.timeout=5000 +test.memory.limit=100