Java容错安全流程统一

wen java案例 2026-06-24 242

本文目录导读：

Java容错安全流程统一

核心容错框架
断路器实现
重试机制
安全验证器
配置管理
使用示例
监控和指标收集
最佳实践建议

我将为您设计一个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";
    }
}

这个统一的容错安全流程框架提供了：