为什么客户端需要负载均衡
服务端负载均衡（Nginx、HAProxy）是常见方案，但响应式系统更倾向于客户端负载均衡：
- 减少网络跳数（直连后端，不经负载均衡器）
- 更细粒度的实例感知（知道哪些实例健康、延迟如何）
- 选择过程本身是异步的，不阻塞调用线程；背压只能约束「单个请求的响应流」，不会自动改变负载均衡的选择
Spring Cloud LoadBalancer 基础
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-loadbalancer</artifactId>
</dependency>
配置服务列表（开发环境）：
spring:
cloud:
loadbalancer:
configurations: default
discovery:
client:
simple:
instances:
order-service:
- uri: http://order-1:8081
metadata:
weight: "70"
- uri: http://order-2:8081
metadata:
weight: "30"
ReactiveLoadBalancer 实现
自定义轮询负载均衡器：
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.DefaultRequest;
import org.springframework.cloud.client.loadbalancer.DefaultResponse;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancer;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
public class RoundRobinLoadBalancer implements ReactiveLoadBalancer<ServiceInstance> {
private final List<ServiceInstance> instances;
private final AtomicInteger position = new AtomicInteger(0);
public RoundRobinLoadBalancer(List<ServiceInstance> instances) {
this.instances = instances;
}
@Override
public Mono<Response<ServiceInstance>> choose(Request request) {
if (instances.isEmpty()) {
return Mono.error(new IllegalStateException("No instances available"));
}
// 计数器溢出后会变成负数，floorMod 保证下标仍落在 [0, size)
int pos = Math.floorMod(position.getAndIncrement(), instances.size());
return Mono.just(new DefaultResponse(instances.get(pos)));
}
}
加权随机策略
public class WeightedRandomLoadBalancer implements ReactiveLoadBalancer<ServiceInstance> {
private final List<ServiceInstance> instances;
public WeightedRandomLoadBalancer(List<ServiceInstance> instances) {
this.instances = instances;
}
@Override
public Mono<Response<ServiceInstance>> choose(Request request) {
if (instances.isEmpty()) {
return Mono.error(new IllegalStateException("No instances available"));
}
int totalWeight = instances.stream()
.mapToInt(i -> Integer.parseInt(i.getMetadata().getOrDefault("weight", "1")))
.sum();
if (totalWeight <= 0) {
return Mono.error(new IllegalStateException("All weights are zero"));
}
// ThreadLocalRandom 是线程本地对象：每次在调用线程上取，不要存成字段跨线程复用
int randomWeight = ThreadLocalRandom.current().nextInt(totalWeight);
int currentWeight = 0;
for (ServiceInstance instance : instances) {
currentWeight += Integer.parseInt(
instance.getMetadata().getOrDefault("weight", "1"));
if (randomWeight < currentWeight) {
return Mono.just(new DefaultResponse(instance));
}
}
return Mono.just(new DefaultResponse(instances.get(instances.size() - 1)));
}
}
与 WebClient 集成
@Service
public class OrderServiceClient {
private final WebClient.Builder webClientBuilder;
private final ReactiveLoadBalancer<ServiceInstance> loadBalancer;
public Mono<String> getOrders() {
return loadBalancer.choose(new DefaultRequest<>())
.flatMap(response -> {
ServiceInstance instance = response.getServer();
String url = instance.getUri().toString() + "/api/orders";
return webClientBuilder.build()
.get().uri(url)
.retrieve()
.bodyToMono(String.class);
});
}
}
健康检查与实例剔除
@Component
public class HealthCheckedLoadBalancer implements ReactiveLoadBalancer<ServiceInstance> {
private final List<ServiceInstance> instances;
private final WebClient healthClient;
private final Map<String, Boolean> healthStatus = new ConcurrentHashMap<>();
public HealthCheckedLoadBalancer(List<ServiceInstance> instances, WebClient healthClient) {
this.instances = instances;
this.healthClient = healthClient;
}
// instanceId 在 SimpleDiscoveryClient 里默认是 null，不能直接当 Map key
private static String keyOf(ServiceInstance instance) {
return instance.getInstanceId() != null
? instance.getInstanceId()
: instance.getUri().toString();
}
// 定时健康检查；需要在配置类上 @EnableScheduling
@Scheduled(fixedRate = 10000)
public void checkHealth() {
instances.forEach(instance -> {
String healthUrl = instance.getUri() + "/health";
healthClient.head().uri(healthUrl)
.retrieve()
.toBodilessEntity() // 消费并释放响应体，避免连接悬挂
.doOnNext(entity -> healthStatus.put(
keyOf(instance), entity.getStatusCode().is2xxSuccessful()))
.onErrorResume(err -> {
healthStatus.put(keyOf(instance), false);
return Mono.empty();
})
.subscribe();
});
}
@Override
public Mono<Response<ServiceInstance>> choose(Request request) {
List<ServiceInstance> healthy = instances.stream()
.filter(i -> healthStatus.getOrDefault(keyOf(i), true))
.toList();
if (healthy.isEmpty()) {
return Mono.error(new IllegalStateException("No healthy instances"));
}
int pos = ThreadLocalRandom.current().nextInt(healthy.size());
return Mono.just(new DefaultResponse(healthy.get(pos)));
}
}
连接池隔离（按需）
默认的全局 ConnectionProvider 已经按远端地址分池，同一实例的连接会复用；只有需要按服务或实例做独立配额、故障隔离、监控标签时，才值得自建池。池的数量会随实例数增长，实例下线时必须 dispose()，否则连接和事件循环资源会泄漏：
public class PooledServiceClient {
private final Map<String, ConnectionProvider> pools = new ConcurrentHashMap<>();
public HttpClient getClient(ServiceInstance instance) {
// instanceId 可能为 null，用 URI 兜底，避免 ConcurrentHashMap 拒绝 null key
String key = instance.getInstanceId() != null
? instance.getInstanceId()
: instance.getUri().toString();
ConnectionProvider pool = pools.computeIfAbsent(key, k ->
ConnectionProvider.builder(k)
.maxConnections(20)
.build());
return HttpClient.create(pool)
.baseUrl(instance.getUri().toString());
}
}
实例 A 的连接池满不会影响实例 B，避免单点故障引发全局问题。注意上面每次 getClient 都会新建 HttpClient，生产实现要把 HttpClient 也缓存起来，否则连接池复用了，客户端对象和事件循环资源却在反复创建。
eventd v2 的内部分发
eventd v2 的内部服务调用使用客户端负载均衡：
gateway → [order-service-1 (weight 70%)]
→ [order-service-2 (weight 30%)]
→ [health: OK / latency < 100ms]
配合第 2.3 节的超时重试，形成一个完整的弹性请求管线。
小结
| 策略 | 特点 | 适用 |
|---|---|---|
| 轮询 | 均匀分发 | 实例性能对等 |
| 随机 | 简单 | 实例性能对等 |
| 加权 | 按权重分发 | 实例性能不等 |
| 健康检查 | 剔除故障实例 | 所有场景 |
| 最少连接 | 选当前连接最少 | 长连接场景 |
下一节 11.2 熔断器 ：Resilience4j CircuitBreakerOperator、状态机与半开恢复。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。