本文目录导读：
我来详细解释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));
}
}
核心要点：
- 最少连接数：动态选择当前连接数最少的服务器
- 指数退避：失败后增加等待时间，避免重复请求
- 抖动(Jitter)：防止惊群效应
- 熔断机制：连续失败后暂时隔离服务器
- 监控：记录连接状态和退避情况
选择哪种策略取决于你的具体场景：对实时性要求高的用最少连接,对稳定性要求高的加入退避机制。
