学习视频 https://www.bilibili.com/video/BV18E411x7eT?p=36

简介

Spring Cloud Ribbon是基于Netflix Ribbon实现的一套客户端负载均衡的工具。
Ribbon 负载均衡的策略包括：

• 轮询（默认）
• 随机
• 加权
  等 …

Ribbon 简单地说就是负载均衡 + RestTemplate 调用，实现RPC的远程调用。

新版的eureka中已经集成了ribbon，所以就另外添加依赖。

一、RestTemplate调用

主要分为get,post两类。
Entity比Object多了一些响应头信息。

• getForObject() / getForEntity()
• postForObject() / postForEntity()

1 getForObject()

@GetMapping("/get/{id}")
    public CommonResult<Payment> getPayment(@PathVariable("id") Long id){
        return restTemplate.getForObject(PAYMENT_URL+"/payment/get/"+id,CommonResult.class);
    }

2 getForEntity()

 @GetMapping("/getEntity/{id}")
    public CommonResult<Payment> getEntity(@PathVariable("id") Long id){
        ResponseEntity<CommonResult>  entity= restTemplate.getForEntity(PAYMENT_URL+"/payment/get/"+id,CommonResult.class);
        if(entity.getStatusCode().is2xxSuccessful()){
            return  entity.getBody();
        }else{
            return new CommonResult<>(444,"存在异常！");
        }
    }

3 postForObject()

@PostMapping("/create")
    public CommonResult<Payment> create(@RequestBody Payment payment){
        log.info("*******消费者启动创建订单*******");
        return restTemplate.postForObject(PAYMENT_URL+"/payment/create",payment, CommonResult.class);
    }

二、修改Ribbon的负载均衡策略

轮询是默认的负载均衡策略，8001，8002两个支付服务会轮流切换。现在我们将轮询改为随机。

1 目录结构

2 创建随机策略类

我们自定义的策略类不能放在@ComponentScan所扫描的当前包及其子包中，不然的话，相当于整个项目都采用了我们定制的策略，就达不到特殊定制的要求了。也就是说我们不能将自定义的策略类放在主启动类的同一级。

MySelfRule类放在新建的 org.example.myrule 目录下。

package org.example.myrule;

import com.netflix.loadbalancer.IRule;
import com.netflix.loadbalancer.RandomRule;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MySelfRule {

    @Bean
    public IRule muRule(){
        return new RandomRule();
    }
}

3 主启动类中添加注解

name指定服务名称，configuration 指定负载均衡的策略。

@RibbonClient(name="CLOUD-PROVIDER-SERVICE",configuration = MySelfRule.class)

4 测试

重新调用 http://localhost/consumer/payment/get/3。

应该可以看到8001与8002不再依次出现了。

三、重写轮询算法

1 Ribbon源码

1）IRule接口

//IRule接口
public interface IRule{
    /*
     * choose one alive server from lb.allServers or
     * lb.upServers according to key
     * 
     * @return choosen Server object. NULL is returned if none
     *  server is available 
     */
	//选择哪个服务实例
    public Server choose(Object key);
    
    public void setLoadBalancer(ILoadBalancer lb);
    
    public ILoadBalancer getLoadBalancer();    
}

2）RoundRobinRule 轮询源码

public class RoundRobinRule extends AbstractLoadBalancerRule {

    private AtomicInteger nextServerCyclicCounter;
    private static final boolean AVAILABLE_ONLY_SERVERS = true;
    private static final boolean ALL_SERVERS = false;

    private static Logger log = LoggerFactory.getLogger(RoundRobinRule.class);

    public RoundRobinRule() {
        nextServerCyclicCounter = new AtomicInteger(0);
    }

    public RoundRobinRule(ILoadBalancer lb) {
        this();
        setLoadBalancer(lb);
    }

    public Server choose(ILoadBalancer lb, Object key) {
        if (lb == null) {
            log.warn("no load balancer");
            return null;
        }

        Server server = null;
        int count = 0;
        while (server == null && count++ < 10) {
            List<Server> reachableServers = lb.getReachableServers();
            List<Server> allServers = lb.getAllServers();
            int upCount = reachableServers.size();
            int serverCount = allServers.size();

            if ((upCount == 0) || (serverCount == 0)) {
                log.warn("No up servers available from load balancer: " + lb);
                return null;
            }

            int nextServerIndex = incrementAndGetModulo(serverCount);
            server = allServers.get(nextServerIndex);

            if (server == null) {
                /* Transient. */
                Thread.yield();
                continue;
            }

            if (server.isAlive() && (server.isReadyToServe())) {
                return (server);
            }

            // Next.
            server = null;
        }

        if (count >= 10) {
            log.warn("No available alive servers after 10 tries from load balancer: "
                    + lb);
        }
        return server;
    }

    /**
     * Inspired by the implementation of {@link AtomicInteger#incrementAndGet()}.
     *
     * @param modulo The modulo to bound the value of the counter.
     * @return The next value.
     */
    private int incrementAndGetModulo(int modulo) {
        for (;;) {
            int current = nextServerCyclicCounter.get();
            int next = (current + 1) % modulo;
            if (nextServerCyclicCounter.compareAndSet(current, next))
                return next;
        }
    }

    @Override
    public Server choose(Object key) {
        return choose(getLoadBalancer(), key);
    }

    @Override
    public void initWithNiwsConfig(IClientConfig clientConfig) {
    }
}

2、重写轮询算法

思路：

• rest 接口第几次请求数 % 服务器集群总数量 = 实际调用服务器位置下标
• 每次服务器重启后rest接口数从1开始
• 接口请求数达到整数最大值时变成0

目录结构如下：

1、在8001和8002支付模块中添加返回端口号的接口

   /**
     * 返回当前模块的端口号信息
     */
    @GetMapping(value = "/lb")
    public String getPaymentLB(){
        return serverPort;
    }

2 注释掉ribbon自带的负载均衡注解@LoadBalanced

3 添加接口LoadBalancer 返回调用的服务信息

package org.example.springcloud.lb;

import org.springframework.cloud.client.ServiceInstance;

import java.util.List;

public interface LoadBalancer {
    ServiceInstance instances(List<ServiceInstance> serviceInstances);
}

4 MyLB 实现 LoadBalancer 接口

package org.example.springcloud.lb;

import org.springframework.cloud.client.ServiceInstance;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class MyLB implements LoadBalancer{
    /**记录当前接口请求的次数*/
    private AtomicInteger atomicInteger=new AtomicInteger(0);

    /**
     * 1、获取当前接口请求次数
     * 2、使用 CAS(比较和替换) 自选锁的方式 ,判断当前值(this.atomicInteger.get())与期望值（current）
     * 是否一致。
     *      若一致：代表当前没有其他用户请求该接口，可以修改 请求接口次数的值 atomicInteger + 1。
     *      若不一致：存在其他用户操作该值，自旋等待直到当前值与期望值一致。
     * @return 修改后的请求接口次数
     */
    private final int getAndIncrement(){
        int current;
        int next;
        do{
            current=this.atomicInteger.get();
            next = current >= Integer.MAX_VALUE ? 0 : current+1;
        }while (!this.atomicInteger.compareAndSet(current,next));
        System.out.println("*****第几次访问，次数next*****:"+next);
        return next;
    }

    /**
     * 实际调用服务器位置下标 = rest 接口第几次请求数 % 服务器集群总数量
     * @param serviceInstances
     * @return 下次服务请求的端口号
     */
    @Override
    public ServiceInstance instances(List<ServiceInstance> serviceInstances) {
        int index = getAndIncrement() % serviceInstances.size();
        return serviceInstances.get(index);
    }
}

5 修改 OrderController

package org.example.springcloud.controller;

import lombok.extern.slf4j.Slf4j;
import org.example.springcloud.entities.CommonResult;
import org.example.springcloud.entities.Payment;
import org.example.springcloud.lb.LoadBalancer;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.net.URI;
import java.util.List;

@RestController
@Slf4j
@RequestMapping("/consumer/payment")
public class OrderController {

    //public static final String PAYMENT_URL= "http://localhost:8001";

    public static final String PAYMENT_URL= "http://CLOUD-PROVIDER-SERVICE";

    @Resource
    private RestTemplate restTemplate;

    /**引入我们自己写的轮询接口*/
    @Resource
    private LoadBalancer loadBalancer;

    /**用来获取服务信息*/
    @Resource
    private DiscoveryClient discoveryClient;

    @PostMapping("/create")
    public CommonResult<Payment> create(@RequestBody Payment payment){
        log.info("*******消费者启动创建订单*******");
        return restTemplate.postForObject(PAYMENT_URL+"/payment/create",payment, CommonResult.class);
    }

    @GetMapping("/get/{id}")
    public CommonResult<Payment> getPayment(@PathVariable("id") Long id){
        return restTemplate.getForObject(PAYMENT_URL+"/payment/get/"+id,CommonResult.class);
    }


    @GetMapping("/getEntity/{id}")
    public CommonResult<Payment> getEntity(@PathVariable("id") Long id){
        ResponseEntity<CommonResult>  entity= restTemplate.getForEntity(PAYMENT_URL+"/payment/get/"+id,CommonResult.class);
        if(entity.getStatusCode().is2xxSuccessful()){
            return  entity.getBody();
        }else{
            return new CommonResult<>(444,"存在异常！");
        }
    }
    @GetMapping("/lb")
    public String getPaymentLB(){
        /**返回服务列表信息*/
        List<ServiceInstance> instances=discoveryClient.getInstances("CLOUD-PROVIDER-SERVICE");
        /**判断服务是否存在*/
        if(instances == null || instances.size()<=0){
            return  null;
        }
        /**获取负载均衡（轮询）后的服务信息*/
        ServiceInstance serviceInstance = loadBalancer.instances(instances);
        /**获取服务地址*/
        URI uri = serviceInstance.getUri();
        /**接口调用*/
        return restTemplate.getForObject(uri+"/payment/lb",String.class);
    }

}

6 测试

接口测试：http://localhost/consumer/payment/lb

可以看到8001与8002交替出现。控制台打印的请求次数信息。

第一种方式 使用配置类

如果这个配置类使用@Configuration注解并写在主启动类能够进行包扫描的范围中，那么调用其他微服务，一律使用这一种负载均衡算法。当然也可以写在主启动所在包的外面

配置类的内容如下：

package com.hs.rule;

import com.netflix.loadbalancer.IRule;
import com.netflix.loadbalancer.RandomRule;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Description:  将ribbon默认的负载均衡方式改为随机的
 * @Author 胡尚
 * @Date: 2022/6/9 11:10
 * @Version 1.0
 */
@Configuration
public class MyRuleConfig {

    @Bean
    public IRule iRule(){
        return new RandomRule();
    }
}

然后在主启动类中指定调用某些服务的时候使用我们配置的负载均衡算法

@SpringBootApplication
// 如果只有一个服务就可以直接使用@RibbonClient注解，而不需要@RibbonClients。
// name是写服务提供方的服务名     configuration是指定我们上面创建的配置类
@RibbonClients(value = {
        @RibbonClient(name = "stock-service", configuration = MyRuleConfig.class)
})
public class OrderApplication {

    public static void main(String[] args) {
        SpringApplication.run(OrderApplication.class, args);
    }
}

第二种方式 使用yaml配置

修改的这种负载均衡算法是使用nacos设置的权重方式

# 使用配置文件的方式修改ribbon默认的负载均衡算法      
stock-service: 
  ribbon:
    NFLoadBalancerRuleClassName: com.alibaba.cloud.nacos.ribbon.NacosRule

接下来进行测试，首先在nacos管理平台上面修改订单服务stock的权重

然后再进行调用，就可以发现测试效果了。

如果要使用自定义的负载均算法，一般就自己创建一个类，然后继承AbstractLoadBalancerRule抽象类，然后写相应的负载均衡逻辑。最后再使用上面的两种方法，要么在配置类中创建该对象并注册进容器，要么在yml文件中指定全路径

一、Ribbon负载均衡

如上图所示，负载均衡就是避免单个服务的实例处理大批量请求而导致其他实例空闲，造成资源浪费。负载均衡分为客户端、服务端的负载均衡，它们最大的区别在于维护服务器的清单保存的位置不同，如：Ribbon属于客户端负载均衡，客户端根据注册中心的服务实例的状态，判定是否访问某个实例。那么判定访问哪个实例，就是负载均衡策略要做的事。Ribbon的负载均衡有三个主要接口：com.netflix.loadbalancer.IPing（服务的检测）、com.netflix.loadbalancer.IRule（定义均衡策略）、com.netflix.loadbalancer.ILoadBalancer（根据IRule请求实例）。

二、Ribbon的负载均衡策略

如上图所示，是IRule的类图。如果需要自定义均衡策略，则继承com.netflix.loadbalancer.AbstractLoadBalancerRule抽象类。如下图所示是Ribbon的7种均衡策略。默认是com.netflix.loadbalancer.ZoneAvoidanceRule（综合判定来选择服务实例）。

三、自定义负载均衡策略

注意：自定义均衡策略，则继承com.netflix.loadbalancer.AbstractLoadBalancerRule抽象类。覆写代码参考该抽象类的实现类。如下代码所示。核心代码是chooseIndex(int reachableCount)方法。

package com.common.instance.gateway.config.ribbon;

import com.log.util.LogUtil;
import com.netflix.client.config.IClientConfig;
import com.netflix.loadbalancer.AbstractLoadBalancerRule;
import com.netflix.loadbalancer.ILoadBalancer;
import com.netflix.loadbalancer.Server;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.List;

/**
 * @description 自定义Ribbon负载均衡策略
 * @author TCM
 * @version 1.0
 * @date 2021/4/1 10:58
 **/
@Configuration
public class MyRibbonBalancerRule extends AbstractLoadBalancerRule {

    // 所有服务实例
    private volatile int total;
    // 实例的索引
    private volatile int index;
    // 可用的所有实例
    List<Server> upList = new ArrayList<>();

    public Server choose(ILoadBalancer lb, Object key) {
        LogUtil.info(String.format("lb=%s,key=%s", lb, key.toString()));
        // 负载均衡器是否为null
        if (lb == null) {
            return null;
        }

        Server server = null;

        while (server == null) {
            // 当前线程是否已中断
            if (Thread.interrupted()) {
                return null;
            }

            // 获取负载均衡器中的所有服务实例
            List<Server> allList = lb.getAllServers();
            int serverCount = allList.size();
            if (serverCount == 0) {
                return null;
            }

            if (total == 0) {
                // 可用的所有实例
                upList = lb.getReachableServers();
            }

            // 修改当前索引
            chooseIndex(lb.getReachableServers().size());
            // 根据索引获取服务
            server = upList.get(index);

            if (server == null) {
                Thread.yield();
                continue;
            }
            if (server.isAlive()) {
                return (server);
            }
            server = null;
            Thread.yield();
        }

        return server;

    }

    // 修改当前索引
    protected void chooseIndex(int reachableCount) {
        if (total < 3) {
            if (upList.size() != reachableCount) {
                index = 0;
            }
            total ++;
        } else {
            total = 0;
            index ++;
            if (index >= reachableCount) {
                index = 0;
            }
        }
    }

    @Override
    public Server choose(Object key) {
        return choose(getLoadBalancer(), key);
    }

    @Override
    public void initWithNiwsConfig(IClientConfig clientConfig) {
    }


}

gateway的yml配置内容：ribbon.NFLoadBalancerRuleClassName: com.common.instance.gateway.config.ribbon.MyRibbonBalancerRule

# hystrix开启
feign:
  hystrix:
    enabled: true
hystrix:
  command:
    default:  # default全局有效，service id指定应用有效
      execution:
        timeout:
          # true则超时根据熔断超时，false则ribbon控制
          enabled: true
        isolation:
          thread:
            timeoutInMilliseconds: 4000 # 断路器超时时间，默认1000ms
# ribbon配置
ribbon:
  OkToRetryOnAllOperations: true # 所有请求重试,默认false
  ReadTimeout: 3000               # 负载均衡超时时间，默认值5000ms
  ConnectTimeout: 1000            # 请求连接的超时时间，默认值2000ms
  MaxAutoRetries: 0               # 当前实例的重试次数，默认0
  MaxAutoRetriesNextServer: 2     # 切换实例的重试次数，默认1
  NFLoadBalancerRuleClassName: com.common.instance.gateway.config.ribbon.MyRibbonBalancerRule

spring:
  cloud:
    gateway:
      # gateway发现该eureka下的所有服务
      discovery:
        locator:
          enabled: false
          lowerCaseServiceId: true
      default-filters:
        - name: Hystrix
          args:
            name: default
            fallbackUri: 'forward:/fallback/global'
      routes:
        - id: ${application.name.instance-demo}
          predicates:
            - Path=/${application.name.instance-demo}/**
          uri: lb://${application.name.instance-demo}

gateway路由接口代码：

package com.common.instance.demo.controller;

import com.common.instance.demo.core.Response;
import com.common.instance.demo.service.RibbonService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @description ribbon负载均衡策略
 * @author TCM
 * @version 1.0
 * @date 2021/3/29 20:30
 **/
@RestController
@RequestMapping("/ribbon")
@Api(tags = "Ribbon测试")
public class RibbonController {

    @Resource
    private RibbonService ribbonService;

    @GetMapping("/rule")
    @ApiOperation("Ribbon策略")
    public Response<String> testRibbonRule(int mills) {
        try {
            Thread.sleep(mills);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String result = ribbonService.testRibbonRule();
        return Response.success(result);
    }

}

package com.common.instance.demo.service.impl;

import com.common.instance.demo.service.RibbonService;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

/**
 * @author TCM
 * @version 1.0
 * @description TODO
 * @date 2021/3/29 20:41
 **/
@Service
public class RibbonServiceImpl implements RibbonService {

    @Value("${spring.cloud.client.ip-address}")
    private String ipAddress;

    @Value("${server.port}")
    private String port;

    @Override
    public String testRibbonRule() {
        return String.format("Ribbon均衡策略测试：主机%s:%s", ipAddress, port);
    }
}

测试结果：

连续3次：发生熔断

{
    "success": false,
    "code": "500",
    "message": "网络发生异常，请稍后重试！",
    "tip": "操作失败",
    "data": null
}

连续3次：9012端口

{
    "success": true,
    "code": "200",
    "message": "操作成功",
    "tip": "操作成功",
    "data": "Ribbon均衡策略测试：主机192.168.1.107:9012"
}

连续3次：9013端口

{
    "success": true,
    "code": "200",
    "message": "操作成功",
    "tip": "操作成功",
    "data": "Ribbon均衡策略测试：主机192.168.1.107:9013"
}