目录一、nacos服务注册和发现的工作机制1、nacos服务器启动2、服务注册到nacos3、微服务宕机4、nacos宕机5、gRPC通信协议二、OpenFeign工作原理三、配置负载均衡1、轮询2、权重配置3、NacosBalancer的算法4、自定义负载均衡算法四、总结我们在springcloud_2中讲了nacos的安装与部署服务注册与用OpenFeign发现服务并且还引入了loadbalance负载均衡。但是我们还没有对服务注册发现的机制进行学习以及负载均衡引入了但是没有去实际的应用。所以本节我们需要深挖底层原理观察http请求分析源代码来说明白这件事。一、nacos服务注册和发现的工作机制我们通过观察nacos.log日志来看整个流程到底发生了什么。1、nacos服务器启动服务器启动nacos会检查有无服务注册到nacos所有类型的长连接数微服务业务sdk客户端长连接数和集群节点互通连接数。2、服务注册到nacos此时我们发现其显示有一个新的连接注册成功包含连接的相关信息和微服务的IP地址连接方式是GRPC长连接来源是sdk这说明这是一个微服务将来用于作为服务提供方提供服务服务发现。然后就是开始检查总的连接数微服务客户端长连接数集群节点互通数。紧接着开始定时心跳检查。微服务客户端开始上报实例健康况状。这一次检查时间为3:05:55本轮周期共进行一次健康检查0个实例健康异常。因为我们目前就只有一个实例说明我们的实例都是正常的。紧接着下一次检查是3:06:01刚好过了5s钟本轮检查结果是所有实例都正常。那么我们可以初步得出一个结论微服务客户端每5s主动上报一次健康状况。3、微服务宕机我们强制结束微服务进程。可以发现目前收不到心跳检测了微服务宕机无法发送心跳包。GRPC长连接断开后实例直接被删除并没有我想如果长连接保持网络波动等因素才会出现如果我们先建立长连接然后启用我们的防火墙出站规则那么连接还在但是心跳包发不出去。双方都认为对方还或者控制台实例一直存在。但其实java终端已经报错连接不上nacos。但是此时微服务仍然可以使用因为微服务会把可能调用的地址在本地缓存起来。就算服务提供方与消费者无法连接到nacos靠着本地缓存仍然可以实现调用。4、nacos宕机我们想如果nacos服务端宕机那么会发生什么这时控制台和java终端都报连接错误但是我们的服务仍然可以调用。这也是得益于本地缓存机制。这就是注册中心必须具备的雪崩防护容灾能力 避免「注册中心单点故障直接导致整个微服务集群全部瘫痪」。5、gRPC通信协议gRPCRemote Procedure Call是由 Google 开发的高性能、通用的开源远程过程调用RPC框架基于HTTP/2协议和Protocol Buffers简称 ProtoBuf序列化协议。它支持多种编程语言包括 Python、Java、Go、C 等适用于构建分布式系统和微服务架构。二、OpenFeign工作原理我们引入OpenFeign依赖后使用FeignClient注解声明我们需要远程调用的接口。那么OpenFeign就起到一个代理的作用识别出微服务名称从nacos拿到服务地址然后拼装完整HTTP报文发出http请求建立TCP连接并返回响应。那么具体流程是怎样的呢EnableFeignClients开启扫描扫描所有FeignClient注解的接口注册FeignClientFactroyBean到spring容器。创建动态代理通过JDK动态代理基于接口生成代理类。调用接口方法实际执行代理对象的invoke()方法。接口调用方法进入代理拦截器。解析路径和参数组装RequestTemplate。此时的url还是服务名没有真实IP。当是被地址是服务名进入负载均衡包装的Client:public feign.Response execute(Request request, Request.Options options) throws IOException { // 构造URL URI originalUri URI.create(request.url()); String serviceId originalUri.getHost(); // 此时host还是服务名 Assert.state(serviceId ! null, Request URI does not contain a valid hostname: originalUri); String hint this.getHint(serviceId); // 构造LoadBalancer请求上下文 DefaultRequestRequestDataContext lbRequest new DefaultRequest(new RequestDataContext(LoadBalancerUtils.buildRequestData(request), hint)); ... // 负载均衡器挑选出需要调用的远程服务实例 ServiceInstance instance this.loadBalancerClient.choose(serviceId, lbRequest); ResponseServiceInstance lbResponse new DefaultResponse(instance); if (instance null) { String message Load balancer does not contain an instance for the service serviceId; if (LOG.isWarnEnabled()) { LOG.warn(message); } ... }根据服务名拿到负载均衡器从下层获取Nacos实例列表。实例来自Nacos本地缓存Nacos宕机也能调用。负载均衡选出一个实例把服务名替换成真实IP端口发送HTTP请求解析结果返回。在ReflectiveFeign.class中public T T newInstance(TargetT target, C requestContext) { ReflectiveFeign.TargetSpecificationVerifier.verify(target); MapMethod, InvocationHandlerFactory.MethodHandler methodToHandler this.targetToHandlersByName.apply(target, requestContext); InvocationHandler handler this.factory.create(target, methodToHandler); T proxy (T)Proxy.newProxyInstance(target.type().getClassLoader(), new Class[]{target.type()}, handler); for(InvocationHandlerFactory.MethodHandler methodHandler : methodToHandler.values()) { if (methodHandler instanceof DefaultMethodHandler) { ((DefaultMethodHandler)methodHandler).bindTo(proxy); } } return proxy; }看到熟悉的Proxy.newProxyInstatance方法里面传入类加载器要实现的接口数组还有控制处理器。由此观之OpenFeign是通过jdk动态代理创建的代理实例。三、配置负载均衡如果我们把同一个微服务部署在多态机器上这样可以减轻单一机器的压力。那么必须设置好负载均衡方式要让他们按照自身不同的配置承当不同的压力。如何配置呢负载均衡组件是一个单独的组件之前我们在使用nacos已经引入过了。dependency groupIdorg.springframework.cloud/groupId artifactIdspring-cloud-starter-loadbalancer/artifactId /dependencyloadbalance和nacos都有自己的负载均衡策略我们当初引入loadbalance后就没有配置过了用的是loadbalance的默认负载策略随机分配。所以要使用nacos的权重负载、同集群优先策略和Nacos流量策略需要手动替换到nacos的负载均衡策略这样控制台配置的全总才会被读取。注意是在消费者端启动负载均衡spring: application: name: user-service cloud: nacos: discovery: server-addr: 192.168.169.177:8848 loadbalancer: nacos: enabled: true1、轮询我们启动必要的服务观察控制台状况目前这些服务都在一个集群下我们多开几个blog-service点击修改选项选择VM Options虚拟机选项这里我们写上该服务的端口。配置-Dserver.port9091然后启动即可然后多次访问这个请求/user/getBlogListById?id1会发现两个博客服务都被调用了证明负载均衡确实起到了作用。进一步发现这两个服务器是轮流处理请求的而非随机。这就是默认情况下的负载均衡配置--轮询。2、权重配置在nacos控制台允许设置服务的权重来控制负载权重越大流量越多权重为0则表示没有流量该服务不会被访问。这样做有利于分配资源避免性能较差的机器承受较多的请求压力。点击博客服务详情找到集群这里有多个实例点击编辑配置权重我们将9090端口的服务权重改为3点击确定后重新访问/user/getBlogListById?id1可以发现9090端口的响应次数大致是9091的三倍3、NacosBalancer的算法除了默认的轮询算法控制台可修改的权重配置还支持自定义负载均衡算法。所谓的负载均衡算法就是从目标服务列表中选中一个来处理用户请求响应用户请求。还记得我们之前开启nacos自己的负载均衡器吗开启之后负载均衡就从LoadBanlancer转变到NacosLoadBanlancer见如下源码负载均衡算法实现了ReactorServiceInstanceLoadBalancerpublic class NacosLoadBalancer implements ReactorServiceInstanceLoadBalancer { private static final Logger log LoggerFactory.getLogger(NacosLoadBalancer.class); private final String serviceId; private ObjectProviderServiceInstanceListSupplier serviceInstanceListSupplierProvider; private final NacosDiscoveryProperties nacosDiscoveryProperties; private static final String IPV4_REGEX ((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})(.((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})){3}; private static final String IPV6_KEY IPv6; public static String ipv6; Autowired private InetIPv6Utils inetIPv6Utils; PostConstruct public void init() { String ip this.nacosDiscoveryProperties.getIp(); if (StringUtils.isNotEmpty(ip)) { ipv6 Pattern.matches(((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})(.((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})){3}, ip) ? (String)this.nacosDiscoveryProperties.getMetadata().get(IPv6) : ip; } else { ipv6 this.inetIPv6Utils.findIPv6Address(); } } private ListServiceInstance filterInstanceByIpType(ListServiceInstance instances) { ...过滤内容 } public NacosLoadBalancer(ObjectProviderServiceInstanceListSupplier serviceInstanceListSupplierProvider, String serviceId, NacosDiscoveryProperties nacosDiscoveryProperties) { this.serviceId serviceId; this.serviceInstanceListSupplierProvider serviceInstanceListSupplierProvider; this.nacosDiscoveryProperties nacosDiscoveryProperties; } public MonoResponseServiceInstance choose(Request request) { ServiceInstanceListSupplier supplier (ServiceInstanceListSupplier)this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new); return supplier.get(request).next().map(this::getInstanceResponse); } private ResponseServiceInstance getInstanceResponse(ListServiceInstance serviceInstances) { if (serviceInstances.isEmpty()) { ... } else { try { String clusterName this.nacosDiscoveryProperties.getClusterName(); ListServiceInstance instancesToChoose serviceInstances; if (StringUtils.isNotBlank(clusterName)) { ListServiceInstance sameClusterInstances (List)serviceInstances.stream().filter((serviceInstance) - { String cluster (String)serviceInstance.getMetadata().get(nacos.cluster); return StringUtils.equals(cluster, clusterName); }).collect(Collectors.toList()); if (!CollectionUtils.isEmpty(sameClusterInstances)) { instancesToChoose sameClusterInstances; } } else { log.warn(A cross-cluster call occursname {}, clusterName {}, instance {}, new Object[]{this.serviceId, clusterName, serviceInstances}); } instancesToChoose this.filterInstanceByIpType(instancesToChoose); ServiceInstance instance NacosBalancer.getHostByRandomWeight3(instancesToChoose); return new DefaultResponse(instance); } catch (Exception e) { log.warn(NacosLoadBalancer error, e); return null; } } } }chose把实例列表取出来交给getInstanceResponse处理。getInstanceResponse()首先去找消费者和生产者是否处于同一集群如果在则不需要跨集群访问。然后就是ServiceInstance instance NacosBalancer.getHostByRandomWeight3(instancesToChoose);这里才是负载均衡算法的实现。public class NacosBalancer extends Balancer { private static final String IPV4_REGEX ((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})(.((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})){3}; private static final String IPV6_KEY IPv6; public static Instance getHostByRandomWeight2(ListInstance instances) { return getHostByRandomWeight(instances); } public static ServiceInstance getHostByRandomWeight3(ListServiceInstance serviceInstances) { ... Instance instance getHostByRandomWeight2(nacosInstance); ... return nacosServiceInstance; } private static void convertIPv4ToIPv6(NacosServiceInstance instance) { ... } }从weight3到wight2再到wight1,这个算法最终在Balancer中实现....public class Balancer { protected static Instance getHostByRandomWeight(ListInstance hosts) { LogUtils.NAMING_LOGGER.debug(entry randomWithWeight); if (hosts ! null hosts.size() ! 0) { LogUtils.NAMING_LOGGER.debug(new Chooser); ListPairInstance hostsWithWeight new ArrayList(); ... 过滤健康实例 for(Instance host : hosts) { if (host.isHealthy()) { hostsWithWeight.add(new Pair(host, host.getWeight())); } } ... return (Instance)vipChooser.randomWithWeight(); } else { LogUtils.NAMING_LOGGER.debug(hosts null || hosts.size() 0); return null; } } public static class RandomByWeight { public static ListInstance selectAll(ServiceInfo serviceInfo) { ListInstance hosts serviceInfo.getHosts(); if (CollectionUtils.isEmpty(hosts)) { throw new IllegalStateException(no host to srv for serviceInfo: serviceInfo.getName()); } else { return hosts; } } public static Instance selectHost(ServiceInfo dom) { 。。。 } } }然后又调用了return (Instance)vipChooser.randomWithWeight();public T randomWithWeight() { ChooserK, T.RefT ref this.ref; double random ThreadLocalRandom.current().nextDouble((double)0.0F, (double)1.0F); int index Arrays.binarySearch(ref.weights, random); if (index 0) { index -index - 1; if (index ref.weights.length random ref.weights[index]) { return (T)ref.items.get(index); } else if (ref.weights.length 0) { throw new IllegalStateException(Cumulative Weight wrong , the array length is equal to 0.); } else { return (T)ref.items.get(ref.items.size() - 1); } } else { return (T)ref.items.get(index); } }终于复出水面了算法的核心思路是先拿到服务列表里的所有权重计算出累计权重数组归一化处理。计算一个随机数[0.0 - 1.0,看这个随机数落在哪个区间。例如原始的权重列表为【127】对应服务列表【A,B,C】归一化并计算出累计权重数组【0.1, 0.3, 1.0】这个累计权重数组里有三个元素。假如生成的随机数是0.5,那么就落在0.3到1.0这个区间直接返回这个区间对应的下标index。这个Index怎么计算出来的呢这里借用了一个对有序数组的二分查找方法int index Arrays.binarySearch(ref.weights, random);主要步骤如下在weights中二分查找random如果找到了直接返回下标如果没找到就先算出这个数应该插入在哪个地方然后返回-(该下标)-1例如对于一个有序数组[0.1, 0.3, 1.0]我们想要查找0.25,者显然是找不到的如果要插入应该在1下标插入那么就返回-1-1 -2注意原数组并没有实际插入这个0.25这个返回值后面我们会用到。没找到是常态找到了是极小概率事件。然后就回到了判断index的逻辑中去如果index 0 则index -index - 1,index变成了1变回去了。那么返回第二个实例。里面还有些兜底措施比如index超出服务列表怎么办返回最后一个实例。所以这就是nacosloadbalancer的负载均衡算法的实质。如果我们要去实现自己的负载均衡算法的话就有一下步骤。如果我们自己去写负载均衡算法就不需要去造轮子了直接实现ReactorServiceInstanceLoadBalancer接口4、自定义负载均衡算法package com.miao.lb; import org.springframework.beans.factory.ObjectProvider; import org.springframework.cloud.client.ServiceInstance; import org.springframework.cloud.client.loadbalancer.DefaultResponse; import org.springframework.cloud.client.loadbalancer.EmptyResponse; import org.springframework.cloud.client.loadbalancer.Request; import org.springframework.cloud.client.loadbalancer.Response; import org.springframework.cloud.loadbalancer.core.ReactorServiceInstanceLoadBalancer; import org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier; import reactor.core.publisher.Mono; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; public class MyRoundRobinLoadBalancer implements ReactorServiceInstanceLoadBalancer { private final String serviceId; private final ObjectProviderServiceInstanceListSupplier supplierProvider; private final AtomicInteger position new AtomicInteger(0); public MyRoundRobinLoadBalancer(ObjectProviderServiceInstanceListSupplier supplierProvider, String serviceId) { this.supplierProvider supplierProvider; this.serviceId serviceId; } Override public MonoResponseServiceInstance choose(Request request) { ServiceInstanceListSupplier supplier supplierProvider.getIfAvailable(); if (supplier null) { return Mono.just(new EmptyResponse()); } return supplier.get(request).next().map(this::getInstanceResponse); } private ResponseServiceInstance getInstanceResponse(ListServiceInstance instances) { if (instances.isEmpty()) { return new EmptyResponse(); } int index Math.floorMod(position.getAndIncrement(), 1这里还可以使用轮询策略即取余instances.size()); return new DefaultResponse(instances.get(index)); } }这里我们就是实现了一个极端的方式使用访问第一个实例。也可以采用轮询。编写配置了把算法注册成Beanpublic class MyLbConfig { Bean public ReactorServiceInstanceLoadBalancer myRoundRobinLoadBalancer( ObjectProviderServiceInstanceListSupplier supplierProvider) { return new MyRoundRobinLoadBalancer(supplierProvider, user-service); } }注意这个config不要加configuration否则注册成全局的会导致这个算法被所有所有场景都应用不同的场景我们希望是不同的算法。然后在消费者启动类中添加使能注解。value就填写需要调用的服务名等价于name填写需要调用的服务名所有实例全部启动后看效果。好像并不是只访问一个固定的实例。因为每次从nacos拉取实例列表表中的顺序是不固定的。想要做到固定需要按照端地址号或者端口号排序。例如启动后看效果符合预期。四、总结这次实验主要是学习了nacos服务注册和发现的简单底层机制以及负载均衡的配置与自定义算法的编写。OpenFeign是通过jdk动态代理实现的而且会把服务列表做本地缓存提升容灾能力。掌握这些知识有助于我们后续为分布式学习打下基础。
springcloud_3:nacos服务注册发现的底层原理和负载均衡
目录一、nacos服务注册和发现的工作机制1、nacos服务器启动2、服务注册到nacos3、微服务宕机4、nacos宕机5、gRPC通信协议二、OpenFeign工作原理三、配置负载均衡1、轮询2、权重配置3、NacosBalancer的算法4、自定义负载均衡算法四、总结我们在springcloud_2中讲了nacos的安装与部署服务注册与用OpenFeign发现服务并且还引入了loadbalance负载均衡。但是我们还没有对服务注册发现的机制进行学习以及负载均衡引入了但是没有去实际的应用。所以本节我们需要深挖底层原理观察http请求分析源代码来说明白这件事。一、nacos服务注册和发现的工作机制我们通过观察nacos.log日志来看整个流程到底发生了什么。1、nacos服务器启动服务器启动nacos会检查有无服务注册到nacos所有类型的长连接数微服务业务sdk客户端长连接数和集群节点互通连接数。2、服务注册到nacos此时我们发现其显示有一个新的连接注册成功包含连接的相关信息和微服务的IP地址连接方式是GRPC长连接来源是sdk这说明这是一个微服务将来用于作为服务提供方提供服务服务发现。然后就是开始检查总的连接数微服务客户端长连接数集群节点互通数。紧接着开始定时心跳检查。微服务客户端开始上报实例健康况状。这一次检查时间为3:05:55本轮周期共进行一次健康检查0个实例健康异常。因为我们目前就只有一个实例说明我们的实例都是正常的。紧接着下一次检查是3:06:01刚好过了5s钟本轮检查结果是所有实例都正常。那么我们可以初步得出一个结论微服务客户端每5s主动上报一次健康状况。3、微服务宕机我们强制结束微服务进程。可以发现目前收不到心跳检测了微服务宕机无法发送心跳包。GRPC长连接断开后实例直接被删除并没有我想如果长连接保持网络波动等因素才会出现如果我们先建立长连接然后启用我们的防火墙出站规则那么连接还在但是心跳包发不出去。双方都认为对方还或者控制台实例一直存在。但其实java终端已经报错连接不上nacos。但是此时微服务仍然可以使用因为微服务会把可能调用的地址在本地缓存起来。就算服务提供方与消费者无法连接到nacos靠着本地缓存仍然可以实现调用。4、nacos宕机我们想如果nacos服务端宕机那么会发生什么这时控制台和java终端都报连接错误但是我们的服务仍然可以调用。这也是得益于本地缓存机制。这就是注册中心必须具备的雪崩防护容灾能力 避免「注册中心单点故障直接导致整个微服务集群全部瘫痪」。5、gRPC通信协议gRPCRemote Procedure Call是由 Google 开发的高性能、通用的开源远程过程调用RPC框架基于HTTP/2协议和Protocol Buffers简称 ProtoBuf序列化协议。它支持多种编程语言包括 Python、Java、Go、C 等适用于构建分布式系统和微服务架构。二、OpenFeign工作原理我们引入OpenFeign依赖后使用FeignClient注解声明我们需要远程调用的接口。那么OpenFeign就起到一个代理的作用识别出微服务名称从nacos拿到服务地址然后拼装完整HTTP报文发出http请求建立TCP连接并返回响应。那么具体流程是怎样的呢EnableFeignClients开启扫描扫描所有FeignClient注解的接口注册FeignClientFactroyBean到spring容器。创建动态代理通过JDK动态代理基于接口生成代理类。调用接口方法实际执行代理对象的invoke()方法。接口调用方法进入代理拦截器。解析路径和参数组装RequestTemplate。此时的url还是服务名没有真实IP。当是被地址是服务名进入负载均衡包装的Client:public feign.Response execute(Request request, Request.Options options) throws IOException { // 构造URL URI originalUri URI.create(request.url()); String serviceId originalUri.getHost(); // 此时host还是服务名 Assert.state(serviceId ! null, Request URI does not contain a valid hostname: originalUri); String hint this.getHint(serviceId); // 构造LoadBalancer请求上下文 DefaultRequestRequestDataContext lbRequest new DefaultRequest(new RequestDataContext(LoadBalancerUtils.buildRequestData(request), hint)); ... // 负载均衡器挑选出需要调用的远程服务实例 ServiceInstance instance this.loadBalancerClient.choose(serviceId, lbRequest); ResponseServiceInstance lbResponse new DefaultResponse(instance); if (instance null) { String message Load balancer does not contain an instance for the service serviceId; if (LOG.isWarnEnabled()) { LOG.warn(message); } ... }根据服务名拿到负载均衡器从下层获取Nacos实例列表。实例来自Nacos本地缓存Nacos宕机也能调用。负载均衡选出一个实例把服务名替换成真实IP端口发送HTTP请求解析结果返回。在ReflectiveFeign.class中public T T newInstance(TargetT target, C requestContext) { ReflectiveFeign.TargetSpecificationVerifier.verify(target); MapMethod, InvocationHandlerFactory.MethodHandler methodToHandler this.targetToHandlersByName.apply(target, requestContext); InvocationHandler handler this.factory.create(target, methodToHandler); T proxy (T)Proxy.newProxyInstance(target.type().getClassLoader(), new Class[]{target.type()}, handler); for(InvocationHandlerFactory.MethodHandler methodHandler : methodToHandler.values()) { if (methodHandler instanceof DefaultMethodHandler) { ((DefaultMethodHandler)methodHandler).bindTo(proxy); } } return proxy; }看到熟悉的Proxy.newProxyInstatance方法里面传入类加载器要实现的接口数组还有控制处理器。由此观之OpenFeign是通过jdk动态代理创建的代理实例。三、配置负载均衡如果我们把同一个微服务部署在多态机器上这样可以减轻单一机器的压力。那么必须设置好负载均衡方式要让他们按照自身不同的配置承当不同的压力。如何配置呢负载均衡组件是一个单独的组件之前我们在使用nacos已经引入过了。dependency groupIdorg.springframework.cloud/groupId artifactIdspring-cloud-starter-loadbalancer/artifactId /dependencyloadbalance和nacos都有自己的负载均衡策略我们当初引入loadbalance后就没有配置过了用的是loadbalance的默认负载策略随机分配。所以要使用nacos的权重负载、同集群优先策略和Nacos流量策略需要手动替换到nacos的负载均衡策略这样控制台配置的全总才会被读取。注意是在消费者端启动负载均衡spring: application: name: user-service cloud: nacos: discovery: server-addr: 192.168.169.177:8848 loadbalancer: nacos: enabled: true1、轮询我们启动必要的服务观察控制台状况目前这些服务都在一个集群下我们多开几个blog-service点击修改选项选择VM Options虚拟机选项这里我们写上该服务的端口。配置-Dserver.port9091然后启动即可然后多次访问这个请求/user/getBlogListById?id1会发现两个博客服务都被调用了证明负载均衡确实起到了作用。进一步发现这两个服务器是轮流处理请求的而非随机。这就是默认情况下的负载均衡配置--轮询。2、权重配置在nacos控制台允许设置服务的权重来控制负载权重越大流量越多权重为0则表示没有流量该服务不会被访问。这样做有利于分配资源避免性能较差的机器承受较多的请求压力。点击博客服务详情找到集群这里有多个实例点击编辑配置权重我们将9090端口的服务权重改为3点击确定后重新访问/user/getBlogListById?id1可以发现9090端口的响应次数大致是9091的三倍3、NacosBalancer的算法除了默认的轮询算法控制台可修改的权重配置还支持自定义负载均衡算法。所谓的负载均衡算法就是从目标服务列表中选中一个来处理用户请求响应用户请求。还记得我们之前开启nacos自己的负载均衡器吗开启之后负载均衡就从LoadBanlancer转变到NacosLoadBanlancer见如下源码负载均衡算法实现了ReactorServiceInstanceLoadBalancerpublic class NacosLoadBalancer implements ReactorServiceInstanceLoadBalancer { private static final Logger log LoggerFactory.getLogger(NacosLoadBalancer.class); private final String serviceId; private ObjectProviderServiceInstanceListSupplier serviceInstanceListSupplierProvider; private final NacosDiscoveryProperties nacosDiscoveryProperties; private static final String IPV4_REGEX ((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})(.((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})){3}; private static final String IPV6_KEY IPv6; public static String ipv6; Autowired private InetIPv6Utils inetIPv6Utils; PostConstruct public void init() { String ip this.nacosDiscoveryProperties.getIp(); if (StringUtils.isNotEmpty(ip)) { ipv6 Pattern.matches(((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})(.((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})){3}, ip) ? (String)this.nacosDiscoveryProperties.getMetadata().get(IPv6) : ip; } else { ipv6 this.inetIPv6Utils.findIPv6Address(); } } private ListServiceInstance filterInstanceByIpType(ListServiceInstance instances) { ...过滤内容 } public NacosLoadBalancer(ObjectProviderServiceInstanceListSupplier serviceInstanceListSupplierProvider, String serviceId, NacosDiscoveryProperties nacosDiscoveryProperties) { this.serviceId serviceId; this.serviceInstanceListSupplierProvider serviceInstanceListSupplierProvider; this.nacosDiscoveryProperties nacosDiscoveryProperties; } public MonoResponseServiceInstance choose(Request request) { ServiceInstanceListSupplier supplier (ServiceInstanceListSupplier)this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new); return supplier.get(request).next().map(this::getInstanceResponse); } private ResponseServiceInstance getInstanceResponse(ListServiceInstance serviceInstances) { if (serviceInstances.isEmpty()) { ... } else { try { String clusterName this.nacosDiscoveryProperties.getClusterName(); ListServiceInstance instancesToChoose serviceInstances; if (StringUtils.isNotBlank(clusterName)) { ListServiceInstance sameClusterInstances (List)serviceInstances.stream().filter((serviceInstance) - { String cluster (String)serviceInstance.getMetadata().get(nacos.cluster); return StringUtils.equals(cluster, clusterName); }).collect(Collectors.toList()); if (!CollectionUtils.isEmpty(sameClusterInstances)) { instancesToChoose sameClusterInstances; } } else { log.warn(A cross-cluster call occursname {}, clusterName {}, instance {}, new Object[]{this.serviceId, clusterName, serviceInstances}); } instancesToChoose this.filterInstanceByIpType(instancesToChoose); ServiceInstance instance NacosBalancer.getHostByRandomWeight3(instancesToChoose); return new DefaultResponse(instance); } catch (Exception e) { log.warn(NacosLoadBalancer error, e); return null; } } } }chose把实例列表取出来交给getInstanceResponse处理。getInstanceResponse()首先去找消费者和生产者是否处于同一集群如果在则不需要跨集群访问。然后就是ServiceInstance instance NacosBalancer.getHostByRandomWeight3(instancesToChoose);这里才是负载均衡算法的实现。public class NacosBalancer extends Balancer { private static final String IPV4_REGEX ((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})(.((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})){3}; private static final String IPV6_KEY IPv6; public static Instance getHostByRandomWeight2(ListInstance instances) { return getHostByRandomWeight(instances); } public static ServiceInstance getHostByRandomWeight3(ListServiceInstance serviceInstances) { ... Instance instance getHostByRandomWeight2(nacosInstance); ... return nacosServiceInstance; } private static void convertIPv4ToIPv6(NacosServiceInstance instance) { ... } }从weight3到wight2再到wight1,这个算法最终在Balancer中实现....public class Balancer { protected static Instance getHostByRandomWeight(ListInstance hosts) { LogUtils.NAMING_LOGGER.debug(entry randomWithWeight); if (hosts ! null hosts.size() ! 0) { LogUtils.NAMING_LOGGER.debug(new Chooser); ListPairInstance hostsWithWeight new ArrayList(); ... 过滤健康实例 for(Instance host : hosts) { if (host.isHealthy()) { hostsWithWeight.add(new Pair(host, host.getWeight())); } } ... return (Instance)vipChooser.randomWithWeight(); } else { LogUtils.NAMING_LOGGER.debug(hosts null || hosts.size() 0); return null; } } public static class RandomByWeight { public static ListInstance selectAll(ServiceInfo serviceInfo) { ListInstance hosts serviceInfo.getHosts(); if (CollectionUtils.isEmpty(hosts)) { throw new IllegalStateException(no host to srv for serviceInfo: serviceInfo.getName()); } else { return hosts; } } public static Instance selectHost(ServiceInfo dom) { 。。。 } } }然后又调用了return (Instance)vipChooser.randomWithWeight();public T randomWithWeight() { ChooserK, T.RefT ref this.ref; double random ThreadLocalRandom.current().nextDouble((double)0.0F, (double)1.0F); int index Arrays.binarySearch(ref.weights, random); if (index 0) { index -index - 1; if (index ref.weights.length random ref.weights[index]) { return (T)ref.items.get(index); } else if (ref.weights.length 0) { throw new IllegalStateException(Cumulative Weight wrong , the array length is equal to 0.); } else { return (T)ref.items.get(ref.items.size() - 1); } } else { return (T)ref.items.get(index); } }终于复出水面了算法的核心思路是先拿到服务列表里的所有权重计算出累计权重数组归一化处理。计算一个随机数[0.0 - 1.0,看这个随机数落在哪个区间。例如原始的权重列表为【127】对应服务列表【A,B,C】归一化并计算出累计权重数组【0.1, 0.3, 1.0】这个累计权重数组里有三个元素。假如生成的随机数是0.5,那么就落在0.3到1.0这个区间直接返回这个区间对应的下标index。这个Index怎么计算出来的呢这里借用了一个对有序数组的二分查找方法int index Arrays.binarySearch(ref.weights, random);主要步骤如下在weights中二分查找random如果找到了直接返回下标如果没找到就先算出这个数应该插入在哪个地方然后返回-(该下标)-1例如对于一个有序数组[0.1, 0.3, 1.0]我们想要查找0.25,者显然是找不到的如果要插入应该在1下标插入那么就返回-1-1 -2注意原数组并没有实际插入这个0.25这个返回值后面我们会用到。没找到是常态找到了是极小概率事件。然后就回到了判断index的逻辑中去如果index 0 则index -index - 1,index变成了1变回去了。那么返回第二个实例。里面还有些兜底措施比如index超出服务列表怎么办返回最后一个实例。所以这就是nacosloadbalancer的负载均衡算法的实质。如果我们要去实现自己的负载均衡算法的话就有一下步骤。如果我们自己去写负载均衡算法就不需要去造轮子了直接实现ReactorServiceInstanceLoadBalancer接口4、自定义负载均衡算法package com.miao.lb; import org.springframework.beans.factory.ObjectProvider; import org.springframework.cloud.client.ServiceInstance; import org.springframework.cloud.client.loadbalancer.DefaultResponse; import org.springframework.cloud.client.loadbalancer.EmptyResponse; import org.springframework.cloud.client.loadbalancer.Request; import org.springframework.cloud.client.loadbalancer.Response; import org.springframework.cloud.loadbalancer.core.ReactorServiceInstanceLoadBalancer; import org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier; import reactor.core.publisher.Mono; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; public class MyRoundRobinLoadBalancer implements ReactorServiceInstanceLoadBalancer { private final String serviceId; private final ObjectProviderServiceInstanceListSupplier supplierProvider; private final AtomicInteger position new AtomicInteger(0); public MyRoundRobinLoadBalancer(ObjectProviderServiceInstanceListSupplier supplierProvider, String serviceId) { this.supplierProvider supplierProvider; this.serviceId serviceId; } Override public MonoResponseServiceInstance choose(Request request) { ServiceInstanceListSupplier supplier supplierProvider.getIfAvailable(); if (supplier null) { return Mono.just(new EmptyResponse()); } return supplier.get(request).next().map(this::getInstanceResponse); } private ResponseServiceInstance getInstanceResponse(ListServiceInstance instances) { if (instances.isEmpty()) { return new EmptyResponse(); } int index Math.floorMod(position.getAndIncrement(), 1这里还可以使用轮询策略即取余instances.size()); return new DefaultResponse(instances.get(index)); } }这里我们就是实现了一个极端的方式使用访问第一个实例。也可以采用轮询。编写配置了把算法注册成Beanpublic class MyLbConfig { Bean public ReactorServiceInstanceLoadBalancer myRoundRobinLoadBalancer( ObjectProviderServiceInstanceListSupplier supplierProvider) { return new MyRoundRobinLoadBalancer(supplierProvider, user-service); } }注意这个config不要加configuration否则注册成全局的会导致这个算法被所有所有场景都应用不同的场景我们希望是不同的算法。然后在消费者启动类中添加使能注解。value就填写需要调用的服务名等价于name填写需要调用的服务名所有实例全部启动后看效果。好像并不是只访问一个固定的实例。因为每次从nacos拉取实例列表表中的顺序是不固定的。想要做到固定需要按照端地址号或者端口号排序。例如启动后看效果符合预期。四、总结这次实验主要是学习了nacos服务注册和发现的简单底层机制以及负载均衡的配置与自定义算法的编写。OpenFeign是通过jdk动态代理实现的而且会把服务列表做本地缓存提升容灾能力。掌握这些知识有助于我们后续为分布式学习打下基础。
