Java分布式数据最少连接退避等怎么最少

wen java案例 2026-07-04 193

本文目录导读：

最少连接数算法实现
退避策略实现
组合实现 - 最少连接 + 退避
使用成熟的框架
最佳实践建议

我来详细解释Java中几种实现最少连接数和退避策略的方案:

最少连接数算法实现

基础版本

public class LeastConnectionBalancer {
    private final Map<String, AtomicInteger> connections = new ConcurrentHashMap<>();
    private final List<String> servers;
    public LeastConnectionBalancer(List<String> servers) {
        this.servers = new ArrayList<>(servers);
        servers.forEach(s -> connections.put(s, new AtomicInteger(0)));
    }
    public String getServer() {
        return servers.stream()
            .min(Comparator.comparingInt(s -> connections.get(s).get()))
            .orElseThrow(() -> new RuntimeException("No servers available"));
    }
    public void incrementConnection(String server) {
        connections.get(server).incrementAndGet();
    }
    public void decrementConnection(String server) {
        connections.get(server).decrementAndGet();
    }
}

带权重的最少连接

public class WeightedLeastConnectionBalancer {
    private final Map<String, ConnectionInfo> serverMap = new ConcurrentHashMap<>();
    static class ConnectionInfo {
        final int weight;
        final AtomicInteger currentConnections = new AtomicInteger(0);
        ConnectionInfo(int weight) {
            this.weight = weight;
        }
        double getEffectiveConnections() {
            return (double) currentConnections.get() / weight;
        }
    }
    public String getServer() {
        return serverMap.entrySet().stream()
            .min(Map.Entry.comparingByValue(
                Comparator.comparingDouble(ci -> ci.getEffectiveConnections())))
            .map(Map.Entry::getKey)
            .orElseThrow(RuntimeException::new);
    }
}

退避策略实现

指数退避

public class ExponentialBackoff {
    private static final long BASE_DELAY = 1000; // 1秒
    private static final long MAX_DELAY = 30000; // 30秒
    private static final double MULTIPLIER = 2.0;
    private static final double JITTER = 0.2; // 20%抖动
    public long getDelay(int retryCount) {
        long delay = (long) (BASE_DELAY * Math.pow(MULTIPLIER, retryCount));
        delay = Math.min(delay, MAX_DELAY);
        // 添加随机抖动，防止惊群效应
        double jitterFactor = 1.0 + (Math.random() - 0.5) * 2 * JITTER;
        delay = (long) (delay * jitterFactor);
        return delay;
    }
    // 使用示例
    public void executeWithBackoff(Runnable task) {
        for (int i = 0; i < 5; i++) {
            try {
                task.run();
                return;
            } catch (Exception e) {
                if (i == 4) throw e;
                long delay = getDelay(i);
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(ie);
                }
            }
        }
    }
}

有界退避

public class BoundedBackoff {
    private static final long MIN_DELAY = 100;
    private static final long MAX_DELAY = 10000;
    private static final int MAX_RETRIES = 5;
    public long getDelay(int retryCount) {
        if (retryCount >= MAX_RETRIES) {
            throw new IllegalStateException("Max retries exceeded");
        }
        // 线性退避
        long delay = MIN_DELAY * (1L << retryCount); // 2^n 增长
        delay = Math.min(delay, MAX_DELAY);
        // 添加随机抖动
        delay += (long) (Math.random() * delay * 0.1);
        return delay;
    }
}

组合实现 - 最少连接 + 退避

public class SmartLoadBalancer {
    private final Map<String, ServerState> servers = new ConcurrentHashMap<>();
    private final ExponentialBackoff backoff = new ExponentialBackoff();
    static class ServerState {
        final AtomicInteger connections = new AtomicInteger(0);
        final AtomicReference<Instant> lastFailure = new AtomicReference<>(Instant.MIN);
        final AtomicInteger consecutiveFailures = new AtomicInteger(0);
    }
    public ServerState selectServer() {
        return servers.entrySet().stream()
            .filter(e -> canAcceptRequest(e.getValue()))
            .min(Comparator.comparingInt(e -> e.getValue().connections.get()))
            .map(Map.Entry::getValue)
            .orElseThrow(() -> new RuntimeException("No available servers"));
    }
    private boolean canAcceptRequest(ServerState state) {
        // 检查是否在退避期
        if (state.consecutiveFailures.get() > 0) {
            long backoffTime = backoff.getDelay(
                Math.min(state.consecutiveFailures.get(), 5));
            Instant retryTime = state.lastFailure.get()
                .plusMillis(backoffTime);
            if (Instant.now().isBefore(retryTime)) {
                return false;
            }
        }
        return true;
    }
    public void onSuccess(String server) {
        ServerState state = servers.get(server);
        state.consecutiveFailures.set(0);
        state.connections.decrementAndGet();
    }
    public void onFailure(String server) {
        ServerState state = servers.get(server);
        state.consecutiveFailures.incrementAndGet();
        state.lastFailure.set(Instant.now());
        state.connections.decrementAndGet();
    }
}

使用成熟的框架

Spring Cloud LoadBalancer

@Configuration
public class LoadBalancerConfig {
    @Bean
    public ReactorLoadBalancer<ServiceInstance> reactorServiceInstanceLoadBalancer(
            Environment environment, 
            LoadBalancerClientFactory loadBalancerClientFactory) {
        String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
        return new RoundRobinLoadBalancer(
            loadBalancerClientFactory.getLazyProvider(name, ServiceInstanceListSupplier.class),
            name);
    }
}

Apache HttpClient 的退避策略

HttpClient client = HttpClientBuilder.create()
    .setRetryHandler((exception, executionCount, context) -> {
        if (executionCount > 3) return false;
        if (exception instanceof ConnectTimeoutException) {
            try {
                Thread.sleep(1000L * executionCount);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return true;
        }
        return false;
    })
    .build();

最佳实践建议

配置化

@ConfigurationProperties(prefix = "loadbalancer")
public class LoadBalancerProperties {
    private long baseDelay = 1000;
    private long maxDelay = 30000;
    private double multiplier = 2.0;
    private int maxRetries = 5;
    private double jitterFactor = 0.2;
    // getters and setters
}

监控和度量

@Component
public class LoadBalancerMetrics {
    private final MeterRegistry registry;
    public LoadBalancerMetrics(MeterRegistry registry) {
        this.registry = registry;
    }
    public void recordConnection(String server, int count) {
        registry.gauge("loadbalancer.connections", 
            Tags.of("server", server), count);
    }
    public void recordBackoff(String server, long delay) {
        registry.timer("loadbalancer.backoff")
            .record(Duration.ofMillis(delay));
    }
}

核心要点：