近期业务大量突增微服务性能优化总结-1.改进客户端负载均衡算法

//一定必须是实现ReactorServiceInstanceLoadBalancer
//而不是ReactorLoadBalancer<ServiceInstance>
//因为注册的时候是ReactorServiceInstanceLoadBalancer
@Log4j2
public class RoundRobinWithRequestSeparatedPositionLoadBalancer implements ReactorServiceInstanceLoadBalancer {
    private final ServiceInstanceListSupplier serviceInstanceListSupplier;
    //每次请求算上重试不会超过1分钟
    //对于超过1分钟的，这种请求肯定比较重，不应该重试
    private final LoadingCache<Long, AtomicInteger> positionCache = Caffeine.newBuilder().expireAfterWrite(1, TimeUnit.MINUTES)
            //随机初始值，防止每次都是从第一个开始调用
            .build(k -> new AtomicInteger(ThreadLocalRandom.current().nextInt(0, 1000)));
    private final String serviceId;
    private final Tracer tracer;
    public RoundRobinWithRequestSeparatedPositionLoadBalancer(ServiceInstanceListSupplier serviceInstanceListSupplier, String serviceId, Tracer tracer) {
        this.serviceInstanceListSupplier = serviceInstanceListSupplier;
        this.serviceId = serviceId;
        this.tracer = tracer;
    }
    //每次重试，其实都会调用这个 choose 方法重新获取一个实例
    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) {
        return serviceInstanceListSupplier.get().next().map(serviceInstances -> getInstanceResponse(serviceInstances));
    }
    private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> serviceInstances) {
        if (serviceInstances.isEmpty()) {
            log.warn("No servers available for service: " + this.serviceId);
            return new EmptyResponse();
        }
        return getInstanceResponseByRoundRobin(serviceInstances);
    }
    private Response<ServiceInstance> getInstanceResponseByRoundRobin(List<ServiceInstance> serviceInstances) {
        if (serviceInstances.isEmpty()) {
            log.warn("No servers available for service: " + this.serviceId);
            return new EmptyResponse();
        }
        //为了解决原始算法不同调用并发可能导致一个请求重试相同的实例
        //从 sleuth 的 Tracer 中获取当前请求的上下文
        Span currentSpan = tracer.currentSpan();
        //如果上下文不存在，则可能不是前端用户请求，而是其他某些机制触发，我们就创建一个新的上下文
        if (currentSpan == null) {
            currentSpan = tracer.newTrace();
        }
        //从请求上下文中获取请求的 traceId，用来唯一标识一个请求
        long l = currentSpan.context().traceId();
        AtomicInteger seed = positionCache.get(l);
        int s = seed.getAndIncrement();
        int pos = s % serviceInstances.size();
        log.info("position {}, seed: {}, instances count: {}", pos, s, serviceInstances.size());
        return new DefaultResponse(serviceInstances.stream()
                //实例返回列表顺序可能不同，为了保持一致，先排序再取
                .sorted(Comparator.comparing(ServiceInstance::getInstanceId))
                .collect(Collectors.toList()).get(pos));
    }
}

<dependency>
    <groupId>io.dropwizard.metrics</groupId>
    <artifactId>metrics-core</artifactId>
</dependency>

@Log4j2
public class ServiceInstanceMetrics {
  private static final String CALLING = "-Calling";
  private static final String FAILED = "-Failed";
  private MetricRegistry metricRegistry;
  ServiceInstanceMetrics() {
  }
  public ServiceInstanceMetrics(MetricRegistry metricRegistry) {
    this.metricRegistry = metricRegistry;
  }
  /**
   * 记录调用实例
   * @param serviceInstance
   */
  public void recordServiceInstanceCall(ServiceInstance serviceInstance) {
    String key = serviceInstance.getHost() + ":" + serviceInstance.getPort();
    metricRegistry.counter(key + CALLING).inc();
  }
  /**
   * 记录调用实例结束
   * @param serviceInstance
   * @param isSuccess 是否成功
   */
  public void recordServiceInstanceCalled(ServiceInstance serviceInstance, boolean isSuccess) {
    String key = serviceInstance.getHost() + ":" + serviceInstance.getPort();
    metricRegistry.counter(key + CALLING).dec();
    if (!isSuccess) {
      //不成功则记录失败
      metricRegistry.meter(key + FAILED).mark();
    }
  }
  /**
   * 获取正在运行的调用次数
   * @param serviceInstance
   * @return
   */
  public long getCalling(ServiceInstance serviceInstance) {
    String key = serviceInstance.getHost() + ":" + serviceInstance.getPort();
    long count = metricRegistry.counter(key + CALLING).getCount();
    log.debug("ServiceInstanceMetrics-getCalling: {} -> {}", key, count);
    return count;
  }
  /**
   * 获取最近一分钟调用失败次数分钟速率，其实是滑动平均数
   * @param serviceInstance
   * @return
   */
  public double getFailedInRecentOneMin(ServiceInstance serviceInstance) {
    String key = serviceInstance.getHost() + ":" + serviceInstance.getPort();
    double rate = metricRegistry.meter(key + FAILED).getOneMinuteRate();
    log.debug("ServiceInstanceMetrics-getFailedInRecentOneMin: {} -> {}", key, rate);
    return rate;
  }
}

private final LoadingCache<Long, Set<String>> calledIpPrefixes = Caffeine.newBuilder()
        .expireAfterAccess(3, TimeUnit.MINUTES)
        .build(k -> Sets.newConcurrentHashSet());
private final String serviceId;
private final Tracer tracer;
private final ServiceInstanceMetrics serviceInstanceMetrics;
//每次重试，其实都会调用这个 choose 方法重新获取一个实例
@Override
public Mono<Response<ServiceInstance>> choose(Request request) {
    Span span = tracer.currentSpan();
    return serviceInstanceListSupplier.get().next()
            .map(serviceInstances -> {
                //保持 span 和调用 choose 的 span 一样
                try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) {
                    return getInstanceResponse(serviceInstances);
                }
            });
}
private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> serviceInstances) {
    if (serviceInstances.isEmpty()) {
        log.warn("No servers available for service: " + this.serviceId);
        return new EmptyResponse();
    }
    //读取 spring-cloud-sleuth 的对于当前请求的链路追踪上下文，获取对应的 traceId
    Span currentSpan = tracer.currentSpan();
    if (currentSpan == null) {
        currentSpan = tracer.newTrace();
    }
    long l = currentSpan.context().traceId();
    return getInstanceResponseByRoundRobin(l, serviceInstances);
}
@VisibleForTesting
public Response<ServiceInstance> getInstanceResponseByRoundRobin(long traceId, List<ServiceInstance> serviceInstances) {
    //首先随机打乱列表中实例的顺序
    Collections.shuffle(serviceInstances);
    //需要先将所有参数缓存起来，否则 comparator 会调用多次，并且可能在排序过程中参数发生改变（针对实例的请求统计数据一直在并发改变）
    Map<ServiceInstance, Integer> used = Maps.newHashMap();
    Map<ServiceInstance, Long> callings = Maps.newHashMap();
    Map<ServiceInstance, Double> failedInRecentOneMin = Maps.newHashMap();
    serviceInstances = serviceInstances.stream().sorted(
            Comparator
                    //之前已经调用过的网段，这里排后面
                    .<ServiceInstance>comparingInt(serviceInstance -> {
                        return used.computeIfAbsent(serviceInstance, k -> {
                            return calledIpPrefixes.get(traceId).stream().anyMatch(prefix -> {
                                return serviceInstance.getHost().contains(prefix);
                            }) ? 1 : 0;
                        });
                    })
                    //当前错误率最少的
                    .thenComparingDouble(serviceInstance -> {
                        return failedInRecentOneMin.computeIfAbsent(serviceInstance, k -> {
                            double value = serviceInstanceMetrics.getFailedInRecentOneMin(serviceInstance);
                            //由于使用的是移动平均值（EMA），需要忽略过小的差异（保留两位小数，不是四舍五入，而是直接舍弃）
                            return ((int) (value * 100)) / 100.0;
                        });
                    })
                    //当前负载请求最少的
                    .thenComparingLong(serviceInstance -> {
                        return callings.computeIfAbsent(serviceInstance, k ->
                                serviceInstanceMetrics.getCalling(serviceInstance)
                        );
                    })
    ).collect(Collectors.toList());
    if (serviceInstances.isEmpty()) {
        log.warn("No servers available for service: " + this.serviceId);
        return new EmptyResponse();
    }
    ServiceInstance serviceInstance = serviceInstances.get(0);
    //记录本次返回的网段
    calledIpPrefixes.get(traceId).add(serviceInstance.getHost().substring(0, serviceInstance.getHost().lastIndexOf(".")));
    //目前记录这个只为了兼容之前的单元测试（调用次数测试）
    positionCache.get(traceId).getAndIncrement();
    return new DefaultResponse(serviceInstance);
}