基于往返延迟的无代理Java gRPC客户端-服务端自定义负载均衡实现

核心逻辑 #

在无代理模式下，自定义负载均衡策略需要兼容GCP Traffic Director的xDS协议，同时从客户端侧收集子通道的往返延迟（RTT）指标，以此作为端点选择的核心依据。

关键实现步骤 #

1. 自定义负载均衡策略类 #

继承AdvancedLoadBalancer（grpc-core 1.54.1推荐使用的新版API，灵活性更强），重写核心方法处理端点更新和子通道状态：

public class RttBasedXdsLoadBalancer extends AdvancedLoadBalancer {
    private final Helper helper;
    private final SubChannelMetrics metrics;
    private List<SubChannel> currentSubChannels = Collections.emptyList();

    public RttBasedXdsLoadBalancer(Helper helper) {
        this.helper = helper;
        this.metrics = new SubChannelMetrics();
    }

    @Override
    public void handleResolvedAddresses(ResolvedAddresses resolvedAddresses) {
        // 处理Traffic Director推送的端点列表，更新可用子通道
        List<EquivalentAddressGroup> addressGroups = resolvedAddresses.getAddresses();
        currentSubChannels = helper.createSubChannels(addressGroups, Attributes.EMPTY);
        // 更新选择器，基于当前RTT数据决策
        helper.updatePicker(new RttBasedPicker(currentSubChannels, metrics));
    }

    @Override
    public void handleSubChannelState(SubChannel subChannel, ConnectivityStateInfo stateInfo) {
        // 监听子通道状态变化，清理失效子通道的指标
        if (stateInfo.getState() == ConnectivityState.SHUTDOWN) {
            metrics.removeSubChannel(subChannel);
            currentSubChannels.remove(subChannel);
            helper.updatePicker(new RttBasedPicker(currentSubChannels, metrics));
        }
    }
}

2. 客户端侧RTT指标收集 #

通过ClientInterceptor拦截请求，记录每个请求的往返时间，并关联到对应的子通道：

public class RttCollectingInterceptor implements ClientInterceptor {
    private final SubChannelMetrics metrics;

    public RttCollectingInterceptor(SubChannelMetrics metrics) {
        this.metrics = metrics;
    }

    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
            MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
        long startTime = System.nanoTime();
        // 从通道属性中获取当前绑定的子通道
        SubChannel subChannel = next.attributes().get(Attributes.Key.create("sub-channel-key"));
        
        return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
            @Override
            public void onMessage(RespT message) {
                long rttNanos = System.nanoTime() - startTime;
                if (subChannel != null) {
                    metrics.recordRtt(subChannel, rttNanos);
                }
                super.onMessage(message);
            }
        };
    }
}

注：需要在负载均衡策略创建子通道时，将子通道引用绑定到通道属性中，确保拦截器能获取到对应关系。

3. RTT驱动的端点选择器 #

实现SubChannelPicker，在pickSubChannel方法中筛选出平均RTT最低的可用子通道：

public class RttBasedPicker extends SubChannelPicker {
    private final List<SubChannel> availableSubChannels;
    private final SubChannelMetrics metrics;

    public RttBasedPicker(List<SubChannel> availableSubChannels, SubChannelMetrics metrics) {
        this.availableSubChannels = availableSubChannels;
        this.metrics = metrics;
    }

    @Override
    public PickResult pickSubChannel(PickSubChannelArgs args) {
        if (availableSubChannels.isEmpty()) {
            return PickResult.withError(Status.UNAVAILABLE);
        }
        // 选择平均RTT最低的子通道，用滑动窗口平均平滑波动
        SubChannel bestChannel = Collections.min(availableSubChannels,
                Comparator.comparingDouble(sc -> metrics.getSlidingWindowAvgRtt(sc)));
        return PickResult.withSubChannel(bestChannel);
    }
}

4. 指标存储与平滑处理 #

自定义SubChannelMetrics类，维护每个子通道的RTT数据，用滑动窗口避免单次波动影响决策：

public class SubChannelMetrics {
    private final Map<SubChannel, Deque<Long>> rttHistory = new ConcurrentHashMap<>();
    private static final int WINDOW_SIZE = 50; // 最近50次请求的平均

    public void recordRtt(SubChannel subChannel, long rttNanos) {
        rttHistory.computeIfAbsent(subChannel, k -> new ArrayDeque<>(WINDOW_SIZE))
                .addLast(rttNanos);
        // 保持窗口大小
        if (rttHistory.get(subChannel).size() > WINDOW_SIZE) {
            rttHistory.get(subChannel).removeFirst();
        }
    }

    public double getSlidingWindowAvgRtt(SubChannel subChannel) {
        Deque<Long> history = rttHistory.get(subChannel);
        if (history == null || history.isEmpty()) {
            return Double.MAX_VALUE; // 无数据时视为高延迟
        }
        return history.stream().mapToLong(Long::longValue).average().orElse(Double.MAX_VALUE);
    }

    public void removeSubChannel(SubChannel subChannel) {
        rttHistory.remove(subChannel);
    }
}

5. 适配xDS与客户端配置 #

实现XdsLoadBalancerProvider，让自定义策略能被xDS客户端加载：

@AutoService(LoadBalancerProvider.class)
public class RttXdsLoadBalancerProvider extends XdsLoadBalancerProvider {
    @Override
    public boolean isAvailable() {
        return true;
    }

    @Override
    public int getPriority() {
        return 5; // 优先级高于默认负载均衡策略
    }

    @Override
    public String getPolicyName() {
        return "rtt-based-xds"; // 自定义策略名称
    }

    @Override
    protected LoadBalancer newLoadBalancer(Helper helper) {
        return new RttBasedXdsLoadBalancer(helper);
    }
}

客户端构建通道时指定策略并注册拦截器：

SubChannelMetrics metrics = new SubChannelMetrics();
ManagedChannel channel = ManagedChannelBuilder.forTarget("xds:///your-service-name")
        .defaultLoadBalancingPolicy("rtt-based-xds")
        .intercept(new RttCollectingInterceptor(metrics))
        .build();

注意事项 #

• 处理端点动态更新：Traffic Director会推送端点变化，需及时同步子通道列表并清理失效节点的指标
• 避免决策震荡：必须用滑动窗口或加权平均平滑RTT数据，防止单次异常延迟导致频繁切换端点
• 兼容xDS协议：确保自定义策略能正确处理LDS/EDS推送的地址和状态，不破坏Traffic Director的流量管控逻辑

内容的提问来源于stack exchange，提问作者Slava Mironov