本文目录导读：
我来为你提供几个Java自愈系统的实践案例,从简单到复杂，每个案例都有完整实现。
简单的故障检测与重启机制
这是一个最基础的自愈系统,通过健康检查自动重启故障服务。
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class BasicSelfHealingSystem {
// 服务状态枚举
public enum ServiceStatus {
HEALTHY, DEGRADED, DOWN
}
// 自愈服务类
public static class HealableService {
private String serviceName;
private volatile boolean running = true;
private AtomicInteger failureCount = new AtomicInteger(0);
public HealableService(String name) {
this.serviceName = name;
}
public boolean healthCheck() {
// 模拟健康检查，80%概率健康
boolean healthy = Math.random() > 0.2;
if (healthy) {
failureCount.set(0);
} else {
failureCount.incrementAndGet();
}
return healthy;
}
public void restart() {
System.out.println("[" + serviceName + "] 正在重启...");
try {
Thread.sleep(2000); // 模拟重启时间
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
running = true;
failureCount.set(0);
System.out.println("[" + serviceName + "] 重启完成");
}
public String getServiceName() {
return serviceName;
}
public int getFailureCount() {
return failureCount.get();
}
}
// 自愈管理器
public static class SelfHealingManager {
private Map<String, HealableService> services = new ConcurrentHashMap<>();
private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
private static final int MAX_FAILURES = 3; // 连续失败次数阈值
public void registerService(HealableService service) {
services.put(service.getServiceName(), service);
}
public void startHealing() {
// 每5秒进行一次健康检查
scheduler.scheduleAtFixedRate(() -> {
for (HealableService service : services.values()) {
boolean healthy = service.healthCheck();
if (!healthy && service.getFailureCount() >= MAX_FAILURES) {
System.out.println("检测到服务 " + service.getServiceName() +
" 连续失败 " + service.getFailureCount() + " 次，触发自愈");
CompletableFuture.runAsync(() -> service.restart());
}
}
}, 0, 5, TimeUnit.SECONDS);
}
public void shutdown() {
scheduler.shutdown();
}
}
// 测试主类
public static void main(String[] args) throws InterruptedException {
SelfHealingManager manager = new SelfHealingManager();
HealableService userService = new HealableService("用户服务");
HealableService orderService = new HealableService("订单服务");
manager.registerService(userService);
manager.registerService(orderService);
manager.startHealing();
// 运行30秒后停止
Thread.sleep(30000);
manager.shutdown();
System.out.println("自愈系统已停止");
}
}
基于断路器的微服务自愈系统
这个案例模拟了真实的微服务场景,包含熔断器和自动恢复机制。
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class CircuitBreakerSelfHealing {
// 断路器状态
public enum CircuitState {
CLOSED, // 正常，允许请求通过
OPEN, // 熔断，直接拒绝请求
HALF_OPEN // 半开，尝试放行少量请求
}
// 微服务定义
public static class MicroService {
private String name;
private String healthEndpoint;
private double failureRate; // 模拟故障率
public MicroService(String name, double failureRate) {
this.name = name;
this.failureRate = failureRate;
}
public String callService() throws Exception {
// 模拟调用
if (Math.random() < failureRate) {
throw new ServiceException(name + " 调用失败");
}
return name + " 响应数据: " + UUID.randomUUID().toString().substring(0, 8);
}
public boolean isHealthy() {
return Math.random() > failureRate;
}
public String getName() {
return name;
}
}
public static class ServiceException extends Exception {
public ServiceException(String message) {
super(message);
}
}
// 断路器组件
public static class CircuitBreaker {
private final String serviceName;
private volatile CircuitState state = CircuitState.CLOSED;
private final int threshold; // 失败阈值
private final long timeout; // 熔断超时时间
private final AtomicInteger failureCount = new AtomicInteger(0);
private volatile long lastFailureTime = 0;
public CircuitBreaker(String serviceName, int threshold, long timeout) {
this.serviceName = serviceName;
this.threshold = threshold;
this.timeout = timeout;
}
public synchronized boolean allowRequest() {
switch (state) {
case CLOSED:
return true;
case OPEN:
if (System.currentTimeMillis() - lastFailureTime >= timeout) {
state = CircuitState.HALF_OPEN;
System.out.println("[" + serviceName + "] 熔断器进入半开状态，开始测试");
return true;
}
return false;
case HALF_OPEN:
return true;
default:
return false;
}
}
public synchronized void onSuccess() {
failureCount.set(0);
if (state == CircuitState.HALF_OPEN) {
state = CircuitState.CLOSED;
System.out.println("[" + serviceName + "] 熔断器关闭，服务恢复正常");
}
}
public synchronized void onFailure() {
failureCount.incrementAndGet();
lastFailureTime = System.currentTimeMillis();
if (state == CircuitState.CLOSED && failureCount.get() >= threshold) {
state = CircuitState.OPEN;
System.out.println("[" + serviceName + "] 熔断器打开，服务进入熔断状态");
} else if (state == CircuitState.HALF_OPEN) {
state = CircuitState.OPEN;
System.out.println("[" + serviceName + "] 服务仍异常，继续熔断");
}
}
public CircuitState getState() {
return state;
}
}
// 自愈管理器（增强版）
public static class AdvancedSelfHealingManager {
private Map<String, MicroService> services;
private Map<String, CircuitBreaker> breakers;
private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(4);
public AdvancedSelfHealingManager() {
this.services = new ConcurrentHashMap<>();
this.breakers = new ConcurrentHashMap<>();
}
public void registerService(MicroService service) {
services.put(service.getName(), service);
breakers.put(service.getName(),
new CircuitBreaker(service.getName(), 5, 10000));
}
public Object callService(String serviceName) {
CircuitBreaker breaker = breakers.get(serviceName);
MicroService service = services.get(serviceName);
if (!breaker.allowRequest()) {
System.out.println(serviceName + " 服务熔断中，请求被拒绝");
return "服务暂不可用，已熔断保护";
}
try {
Object result = service.callService();
breaker.onSuccess();
return result;
} catch (Exception e) {
breaker.onFailure();
return "服务调用失败: " + e.getMessage();
}
}
public void startMonitoring() {
// 定时检查服务健康状态
scheduler.scheduleAtFixedRate(() -> {
for (MicroService service : services.values()) {
CircuitBreaker breaker = breakers.get(service.getName());
// 如果服务健康且熔断器打开，尝试恢复
if (service.isHealthy() && breaker.getState() == CircuitState.OPEN) {
breaker.allowRequest(); // 触发半开
System.out.println("[" + service.getName() + "] 检测到服务可能恢复，准备重试");
}
}
}, 0, 10, TimeUnit.SECONDS);
}
public void displayStatus() {
for (Map.Entry<String, CircuitBreaker> entry : breakers.entrySet()) {
System.out.println(entry.getKey() + " 状态: " + entry.getValue().getState());
}
}
}
// 测试
public static void main(String[] args) throws Exception {
AdvancedSelfHealingManager manager = new AdvancedSelfHealingManager();
// 注册两个服务，一个故障率高，一个低
MicroService orderService = new MicroService("订单服务", 0.3);
MicroService paymentService = new MicroService("支付服务", 0.7);
manager.registerService(orderService);
manager.registerService(paymentService);
manager.startMonitoring();
// 模拟40次服务调用
for (int i = 0; i < 40; i++) {
System.out.println("\n调用 " + (i+1) + ":");
System.out.println(manager.callService("订单服务"));
System.out.println(manager.callService("支付服务"));
Thread.sleep(500);
}
System.out.println("\n最终状态:");
manager.displayStatus();
}
}
完整的生产级自愈系统
包含健康检查、自动扩展、日志分析、告警通知等完整功能。
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
public class ProductionSelfHealingSystem {
// 服务节点
public static class ServiceNode {
private String nodeId;
private String serviceName;
private String ip;
private int port;
private volatile boolean healthy = true;
private int healthScore = 100;
private List<String> recentErrors = new ArrayList<>();
private long lastHeartbeat;
public ServiceNode(String nodeId, String serviceName, String ip, int port) {
this.nodeId = nodeId;
this.serviceName = serviceName;
this.ip = ip;
this.port = port;
updateHeartbeat();
}
public void updateHeartbeat() {
this.lastHeartbeat = System.currentTimeMillis();
}
public double getHealthScore() {
return healthScore;
}
public void setHealthScore(int healthScore) {
this.healthScore = Math.max(0, Math.min(100, healthScore));
}
public String getNodeId() { return nodeId; }
public String getServiceName() { return serviceName; }
public String getIp() { return ip; }
public int getPort() { return port; }
public long getLastHeartbeat() { return lastHeartbeat; }
public boolean isHealthy() { return healthy; }
public void setHealthy(boolean healthy) {
this.healthy = healthy;
}
public void addError(String error) {
recentErrors.add(error);
if (recentErrors.size() > 10) {
recentErrors.remove(0);
}
}
public List<String> getRecentErrors() {
return recentErrors;
}
}
// 告警信息
public static class Alert {
private String alertType;
private String message;
private LocalDateTime timestamp;
private AlertLevel level;
public enum AlertLevel {
INFO, WARNING, CRITICAL
}
public Alert(String alertType, String message, AlertLevel level) {
this.alertType = alertType;
this.message = message;
this.level = level;
this.timestamp = LocalDateTime.now();
}
@Override
public String toString() {
return String.format("[%s] [%s] %s: %s",
timestamp, level, alertType, message);
}
}
// 内存管理和监控
public static class ResourceMonitor {
private static final double MEM_THRESHOLD = 0.85; // 内存使用率阈值
private static final double CPU_THRESHOLD = 0.80;
public ResourceStatus checkResources() {
ResourceStatus status = new ResourceStatus();
// 获取JVM内存信息
Runtime runtime = Runtime.getRuntime();
long usedMem = runtime.totalMemory() - runtime.freeMemory();
long maxMem = runtime.maxMemory();
status.memoryUsage = (double) usedMem / maxMem;
// 获取CPU使用率
com.sun.management.OperatingSystemMXBean osBean
= (com.sun.management.OperatingSystemMXBean)ManagementFactory
.getOperatingSystemMXBean();
status.cpuUsage = osBean.getSystemCpuLoad();
// 检查线程数
status.threadCount = Thread.activeCount();
return status;
}
public static class ResourceStatus {
public double memoryUsage;
public double cpuUsage;
public int threadCount;
public boolean hasMemoryProblem() {
return memoryUsage > MEM_THRESHOLD;
}
public boolean hasCpuProblem() {
return cpuUsage > CPU_THRESHOLD;
}
}
}
// 自愈策略
public static class HealingStrategy {
private String strategyName;
private int priority;
private Action action;
public interface Action {
void execute(ServiceNode node);
}
public HealingStrategy(String name, int priority, Action action) {
this.strategyName = name;
this.priority = priority;
this.action = action;
}
public void apply(ServiceNode node) {
System.out.println("执行自愈策略: " + strategyName + " -> " + node.getServiceName());
action.execute(node);
}
}
// 核心自愈系统
public static class ProductionSelfHealingManager {
private Map<String, ServiceNode> serviceRegistry = new ConcurrentHashMap<>();
private List<HealingStrategy> strategies = new ArrayList<>();
private com.uss.alert.AlertManager alertManager = new BaseAlertManager();
private ResourceMonitor resourceMonitor = new ResourceMonitor();
private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(4);
public ProductionSelfHealingManager() {
initializeStrategies();
}
private void initializeStrategies() {
// 策略1: 节点重启
strategies.add(new HealingStrategy("节点重启", 1, (node) -> {
System.out.println("重启节点: " + node.getNodeId());
node.setHealthy(true);
node.setHealthScore(100);
}));
// 策略2: 资源清理
strategies.add(new HealingStrategy("清理内存/GC", 2, (node) -> {
System.out.println("清理节点内存: " + node.getNodeId());
System.gc();
node.setHealthScore(node.getHealthScore() + 20);
}));
// 策略3: 流量转移
strategies.add(new HealingStrategy("流量转移", 3, (node) -> {
System.out.println("转移节点流量: " + node.getNodeId());
// 实际场景: 将流量转移到其他健康节点
node.setHealthScore(0);
}));
}
public void registerServiceNode(ServiceNode node) {
serviceRegistry.put(node.getNodeId(), node);
System.out.println("注册服务节点: " + node.getServiceName() +
" (" + node.getIp() + ":" + node.getPort() + ")");
}
public void startSelfHealing() {
// 定时健康检查
scheduler.scheduleAtFixedRate(() -> {
performHealthChecks();
}, 0, 10, TimeUnit.SECONDS);
// 定时资源监控
scheduler.scheduleAtFixedRate(() -> {
monitorResources();
}, 0, 30, TimeUnit.SECONDS);
// 定时告警检查
scheduler.scheduleAtFixedRate(() -> {
checkAlerts();
}, 0, 5, TimeUnit.SECONDS);
}
private void performHealthChecks() {
for (ServiceNode node : serviceRegistry.values()) {
// 模拟健康状态检测
double random = Math.random();
node.updateHeartbeat();
if (random < 0.3) { // 30%几率出现健康问题
node.setHealthScore(node.getHealthScore() - 15);
node.addError("健康检查失败 - " + UUID.randomUUID().toString().substring(0, 6));
alertManager.sendAlert(new Alert(
"HEALTH_CHECK",
node.getServiceName() + " 健康评分下降至 " + node.getHealthScore(),
Alert.AlertLevel.WARNING
));
if (node.getHealthScore() < 30) { // 健康评分低于30，触发自愈
for (HealingStrategy strategy : strategies) {
strategy.apply(node);
}
}
} else {
// 正常恢复
node.setHealthScore(Math.min(100, node.getHealthScore() + 5));
node.setHealthy(true);
}
}
}
private void monitorResources() {
ResourceMonitor.ResourceStatus status = resourceMonitor.checkResources();
if (status.hasMemoryProblem()) {
alertManager.sendAlert(new Alert(
"RESOURCE",
"系统内存使用率过高: " + String.format("%.2f%%", status.memoryUsage * 100),
Alert.AlertLevel.CRITICAL
));
// 触发内存清理策略
serviceRegistry.values().forEach(node -> {
if (node.getHealthScore() > 50) {
new HealingStrategy("内存回收", 2, (n) -> {
System.out.println("内存回收: " + n.getNodeId());
}).apply(node);
}
});
}
if (status.hasCpuProblem()) {
alertManager.sendAlert(new Alert(
"RESOURCE",
"CPU使用率过高: " + String.format("%.2f%%", status.cpuUsage * 100),
Alert.AlertLevel.WARNING
));
}
}
private void checkAlerts() {
// 检查是否有过期的服务节点
long currentTime = System.currentTimeMillis();
serviceRegistry.forEach((id, node) -> {
if (currentTime - node.getLastHeartbeat() > 30000) {
alertManager.sendAlert(new Alert(
"TIMEOUT",
"节点心跳超时: " + node.getServiceName(),
Alert.AlertLevel.CRITICAL
));
// 自动重启超时节点
node.setHealthy(false);
strategies.get(0).apply(node); // 执行重启策略
}
});
}
public void displaySystemStatus() {
System.out.println("\n=== 系统当前状态 ===");
serviceRegistry.forEach((id, node) -> {
System.out.println(node.getServiceName() +
" [Score: " + String.format("%.0f", node.getHealthScore()) +
"] [Status: " + (node.isHealthy() ? "健康" : "异常") + "]");
});
}
public void shutdown() {
scheduler.shutdown();
}
}
// 告警管理器接口
public interface AlertManager {
void sendAlert(Alert alert);
}
public static class BaseAlertManager implements AlertManager {
@Override
public void sendAlert(Alert alert) {
if (alert.level == Alert.AlertLevel.CRITICAL) {
System.out.println("【紧急告警】" + alert.message);
} else if (alert.level == Alert.AlertLevel.WARNING) {
System.out.println("【警告】" + alert.message);
} else {
System.out.println("[通知]" + alert.message);
}
}
}
// 主测试
public static void main(String[] args) throws InterruptedException, Exception {
ProductionSelfHealingManager manager = new ProductionSelfHealingManager();
// 注册3个服务节点
ServiceNode node1 = new ServiceNode("node-1", "用户服务", "192.168.1.101", 8081);
ServiceNode node2 = new ServiceNode("node-2", "订单服务", "192.168.1.102", 8082);
ServiceNode node3 = new ServiceNode("node-3", "支付服务", "192.168.1.103", 8083);
manager.registerServiceNode(node1);
manager.registerServiceNode(node2);
manager.registerServiceNode(node3);
// 启动自愈系统
manager.startSelfHealing();
System.out.println("自愈系统启动成功");
// 运行60秒后关闭
Thread.sleep(60000);
manager.shutdown();
System.out.println("自愈系统已关闭");
}
}
生产实践要点
检测维度
- 健康检查: 定时探测服务可用性
- 性能指标: CPU、内存、响应时间
- 日志分析: 错误日志、异常模式
自愈策略
// 分层自愈策略 1. L1 - 快速恢复: 重启、重试、降级 2. L2 - 资源调整: 扩容、缩容、负载均衡 3. L3 - 架构调整: 服务转移、流量切换
注意事项
- 避免自愈风暴: 限制自愈操作频率
- 人工干预优先: 对重要操作保留人工确认
- 审计日志: 记录所有自愈动作
- 回滚机制: 确保操作可逆
最佳实践建议
# 自愈配置建议 self-healing: check-interval: 5s # 检测间隔 failure-threshold: 3 # 失败阈值 action-timeout: 30s # 操作超时 retry-count: 2 # 重试次数 cooldown-period: 60s # 冷却时间 alert-enabled: true # 是否启用告警 rollback-enabled: true # 是否启用回滚
这些案例覆盖了从简单到复杂的自愈系统实现,实际生产环境需要根据具体场景调整和扩展。
