本文目录导读：
我将为您设计一个Java容错安全流程的统一框架，包含核心组件、容错策略和安全机制。
核心容错框架
package com.framework.resilience;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.function.Supplier;
/**
* 统一容错安全流程框架
*/
public class ResilientFlow<T, R> {
private final FlowConfig config;
private final CircuitBreaker circuitBreaker;
private final RetryHandler retryHandler;
private final RateLimiter rateLimiter;
private final Bulkhead bulkhead;
private final SecurityValidator securityValidator;
public ResilientFlow(FlowConfig config) {
this.config = config;
this.circuitBreaker = new CircuitBreaker(config);
this.retryHandler = new RetryHandler(config);
this.rateLimiter = new RateLimiter(config);
this.bulkhead = new Bulkhead(config);
this.securityValidator = new SecurityValidator(config);
}
// 执行流程
public R execute(Supplier<R> supplier) {
return executeWithContext(supplier, null);
}
public R execute(Supplier<R> supplier, Function<Throwable, R> fallback) {
try {
// 1. 安全检查
securityValidator.validate();
// 2. 断路器检查
circuitBreaker.checkState();
// 3. 限流检查
rateLimiter.acquire();
// 4. 舱壁隔离
return bulkhead.execute(() -> {
// 5. 重试机制
return retryHandler.execute(supplier);
});
} catch (Exception e) {
// 6. 降级处理
return handleFailure(e, fallback);
}
}
private R executeWithContext(Supplier<R> supplier,
Function<Throwable, R> fallback) {
FlowContext context = new FlowContext();
context.start();
try {
R result = execute(supplier, fallback);
context.success();
return result;
} catch (Exception e) {
context.failure(e);
throw e;
} finally {
context.end();
MetricsCollector.record(context);
}
}
private R handleFailure(Throwable e, Function<Throwable, R> fallback) {
circuitBreaker.recordFailure();
if (fallback != null) {
try {
return fallback.apply(e);
} catch (Exception fallbackException) {
throw new ResilientException("Fallback failed", fallbackException);
}
}
throw new ResilientException("Flow execution failed", e);
}
}
断路器实现
package com.framework.resilience;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
/**
* 断路器模式实现
*/
public class CircuitBreaker {
private enum State { CLOSED, OPEN, HALF_OPEN }
private final FlowConfig config;
private volatile State state = State.CLOSED;
private final AtomicInteger failureCount = new AtomicInteger(0);
private final AtomicInteger successCount = new AtomicInteger(0);
private volatile long lastFailureTime = 0;
public CircuitBreaker(FlowConfig config) {
this.config = config;
}
public synchronized void checkState() {
if (state == State.OPEN) {
if (System.currentTimeMillis() - lastFailureTime > config.getTimeoutMs()) {
state = State.HALF_OPEN;
} else {
throw new CircuitBreakerOpenException("Circuit breaker is open");
}
}
}
public synchronized void recordFailure() {
failureCount.incrementAndGet();
lastFailureTime = System.currentTimeMillis();
if (failureCount.get() >= config.getFailureThreshold()) {
state = State.OPEN;
}
}
public synchronized void recordSuccess() {
if (state == State.HALF_OPEN) {
successCount.incrementAndGet();
if (successCount.get() >= config.getSuccessThreshold()) {
reset();
}
}
}
private void reset() {
state = State.CLOSED;
failureCount.set(0);
successCount.set(0);
}
}
重试机制
package com.framework.resilience;
import java.util.function.Supplier;
/**
* 重试处理器
*/
public class RetryHandler {
private final FlowConfig config;
public RetryHandler(FlowConfig config) {
this.config = config;
}
public <R> R execute(Supplier<R> supplier) {
Exception lastException = null;
for (int attempt = 0; attempt <= config.getMaxRetries(); attempt++) {
try {
R result = supplier.get();
return result;
} catch (Exception e) {
lastException = e;
if (attempt < config.getMaxRetries() && shouldRetry(e)) {
waitBeforeRetry(attempt);
} else {
throw new RuntimeException("All retries failed", e);
}
}
}
throw new RuntimeException("Retry exhausted", lastException);
}
private boolean shouldRetry(Exception e) {
// 根据异常类型决定是否重试
return e instanceof RetryableException
|| e instanceof java.net.ConnectException
|| e instanceof java.io.IOException;
}
private void waitBeforeRetry(int attempt) {
long waitTime = calculateWaitTime(attempt);
try {
TimeUnit.MILLISECONDS.sleep(waitTime);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new RuntimeException("Retry interrupted", ie);
}
}
private long calculateWaitTime(int attempt) {
// 指数退避 + 随机抖动
long baseDelay = config.getRetryDelayMs();
long exponentialDelay = baseDelay * (long) Math.pow(2, attempt);
long jitter = (long) (Math.random() * baseDelay);
return Math.min(exponentialDelay + jitter, config.getMaxRetryDelayMs());
}
}
安全验证器
package com.framework.resilience;
import java.util.concurrent.ConcurrentHashMap;
/**
* 安全验证器
*/
public class SecurityValidator {
private final FlowConfig config;
private final ConcurrentHashMap<String, TokenInfo> tokenCache = new ConcurrentHashMap<>();
public SecurityValidator(FlowConfig config) {
this.config = config;
}
public void validate() {
// 1. 获取当前请求上下文
RequestContext context = RequestContextHolder.get();
// 2. 验证身份
if (!authenticate(context)) {
throw new AuthenticationException("Authentication failed");
}
// 3. 验证权限
if (!authorize(context)) {
throw new AuthorizationException("Insufficient permissions");
}
// 4. 验证输入
if (!validateInput(context)) {
throw new ValidationException("Input validation failed");
}
// 5. 检查是否在黑名单中
if (isBlocked(context)) {
throw new BlockedException("Client is blocked");
}
}
private boolean authenticate(RequestContext context) {
String token = context.getToken();
if (token == null || token.isEmpty()) {
return false;
}
// 验证JWT或Session
return tokenCache.computeIfAbsent(token, this::verifyToken)
.isValid();
}
private TokenInfo verifyToken(String token) {
// JWT验证逻辑
try {
// 解析和验证JWT
return new TokenInfo(true, System.currentTimeMillis() + 3600000);
} catch (Exception e) {
return new TokenInfo(false, 0);
}
}
private boolean authorize(RequestContext context) {
// RBAC权限检查
String role = context.getRole();
String resource = context.getResource();
String action = context.getAction();
return PermissionManager.checkPermission(role, resource, action);
}
private boolean validateInput(RequestContext context) {
// 输入验证
Object payload = context.getPayload();
return InputValidator.validate(payload);
}
private boolean isBlocked(RequestContext context) {
String clientId = context.getClientId();
return BlacklistManager.isBlacklisted(clientId);
}
}
配置管理
package com.framework.resilience;
import lombok.Builder;
import lombok.Data;
/**
* 流程配置
*/
@Data
@Builder
public class FlowConfig {
// 断路器配置
private int failureThreshold = 5; // 失败阈值
private int successThreshold = 3; // 成功阈值
private long timeoutMs = 30000; // 超时时间
// 重试配置
private int maxRetries = 3; // 最大重试次数
private long retryDelayMs = 1000; // 重试延迟
private long maxRetryDelayMs = 30000; // 最大重试延迟
// 限流配置
private int maxRequestsPerSecond = 100; // QPS限制
private int maxConcurrentRequests = 50; // 并发限制
// 舱壁配置
private int maxThreadPoolSize = 10; // 线程池大小
private int maxQueueSize = 100; // 队列大小
// 安全管理
private boolean enableAuthentication = true;
private boolean enableAuthorization = true;
private boolean enableInputValidation = true;
private boolean enableBlacklist = true;
}
使用示例
package com.example;
import com.framework.resilience.*;
import java.util.function.Supplier;
public class Application {
public static void main(String[] args) {
// 创建配置
FlowConfig config = FlowConfig.builder()
.failureThreshold(5)
.maxRetries(3)
.retryDelayMs(1000)
.maxRequestsPerSecond(100)
.build();
// 创建容错流程
ResilientFlow<String, String> flow = new ResilientFlow<>(config);
// 正常执行
Supplier<String> task = () -> callExternalService();
// 带降级的执行
String result = flow.execute(task, exception -> {
return getFallbackValue();
});
System.out.println("Result: " + result);
// 批量执行
for (int i = 0; i < 100; i++) {
int requestId = i;
CompletableFuture.runAsync(() -> {
try {
String response = flow.execute(() -> processRequest(requestId));
System.out.println("Request " + requestId + ": " + response);
} catch (Exception e) {
System.err.println("Request " + requestId + " failed: " + e.getMessage());
}
});
}
}
private static String callExternalService() {
// 模拟外部服务调用
if (Math.random() < 0.3) {
throw new RuntimeException("Service unavailable");
}
return "Success";
}
private static String getFallbackValue() {
return "Fallback response";
}
private static String processRequest(int id) {
// 业务流程
return "Processed: " + id;
}
}
监控和指标收集
package com.framework.resilience;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
/**
* 指标收集器
*/
public class MetricsCollector {
private static final ConcurrentHashMap<String, MetricData> metrics = new ConcurrentHashMap<>();
public static void record(FlowContext context) {
String flowName = context.getFlowName();
MetricData data = metrics.computeIfAbsent(flowName, k -> new MetricData());
data.totalRequests.incrementAndGet();
if (context.isSuccess()) {
data.successRequests.incrementAndGet();
} else {
data.failureRequests.incrementAndGet();
}
data.totalLatency.addAndGet(context.getLatency());
data.lastLatency.set(context.getLatency());
// 更新延迟分布
updateLatencyDistribution(data, context.getLatency());
}
private static void updateLatencyDistribution(MetricData data, long latency) {
if (latency < 100) {
data.latencyBelow100ms.incrementAndGet();
} else if (latency < 500) {
data.latencyBelow500ms.incrementAndGet();
} else if (latency < 1000) {
data.latencyBelow1000ms.incrementAndGet();
} else {
data.latencyAbove1000ms.incrementAndGet();
}
}
public static MetricData getMetrics(String flowName) {
return metrics.getOrDefault(flowName, new MetricData());
}
public static class MetricData {
private AtomicLong totalRequests = new AtomicLong(0);
private AtomicLong successRequests = new AtomicLong(0);
private AtomicLong failureRequests = new AtomicLong(0);
private AtomicLong totalLatency = new AtomicLong(0);
private AtomicLong lastLatency = new AtomicLong(0);
private AtomicLong latencyBelow100ms = new AtomicLong(0);
private AtomicLong latencyBelow500ms = new AtomicLong(0);
private AtomicLong latencyBelow1000ms = new AtomicLong(0);
private AtomicLong latencyAbove1000ms = new AtomicLong(0);
// Getters and statistics methods
public double getSuccessRate() {
long total = totalRequests.get();
return total > 0 ? (double) successRequests.get() / total : 0;
}
public double getAverageLatency() {
long total = totalRequests.get();
return total > 0 ? (double) totalLatency.get() / total : 0;
}
}
}
最佳实践建议
1 配置优化
# application.yml
resilience:
flow:
default:
failure-threshold: 10
max-retries: 3
retry-delay-ms: 500
max-requests-per-second: 200
max-concurrent-requests: 50
critical-service:
failure-threshold: 5
max-retries: 5
max-concurrent-requests: 30
2 异常处理策略
// 自定义异常继承体系
public class ResilientBaseException extends RuntimeException {
private final String errorCode;
private final ErrorSeverity severity;
}
public class RetryableException extends ResilientBaseException {
// 可重试的异常
}
public class NonRetryableException extends ResilientBaseException {
// 不可重试的异常
}
public class CircuitBreakerOpenException extends ResilientBaseException {
// 断路器打开异常
}
3 健康检查
@Component
public class HealthCheckEndpoints {
@GetMapping("/health/resilience")
public ResilienceHealth health() {
MetricData metrics = MetricsCollector.getMetrics("default");
return ResilienceHealth.builder()
.status(getOverallStatus(metrics))
.successRate(metrics.getSuccessRate())
.averageLatency(metrics.getAverageLatency())
.circuitBreakerState(circuitBreaker.getState())
.build();
}
private String getOverallStatus(MetricData metrics) {
if (metrics.getSuccessRate() < 0.9) return "DEGRADED";
if (metrics.getSuccessRate() < 0.5) return "CRITICAL";
return "HEALTHY";
}
}
这个统一的容错安全流程框架提供了：
- 断路器模式
- 重试机制（指数退避）
- 限流控制
- 舱壁隔离
- 安全验证
- 降级处理
- 指标监控
可以根据具体业务需求调整配置参数和扩展实现。
