8.Ribbon负载均衡服务调用

学习B站尚硅谷周阳老师SpringCloud的听课笔记

1.Ribbon概述

1.1是什么？

官网：https://github.com/Netflix/ribbon/wiki/Getting-Started
Ribbon目前也进入维护模式

1.2能干嘛？

LB（负载均衡）
集中式LB
进程内LB
负载均衡+RestTemplate调用

1.3架构说明

总结：Ribbon其实就是一个软负载均衡的客户端组件，他可以和其他所需请求的客户端结合使用，和eureka结合只是其中的一个实例。

1.4Eureka jar包包含Ribbon

1.5二说RestTemplate的使用

官网：https://docs.spring.io/spring-framework/docs/5.2.2.RELEASE/javadoc-api/org/springframework/web/client/RestTemplate.html

ForObject方法/ForEntity方法

2.Ribbon核心组件IRule

2.1 IRule:根据特定算法从服务列表中选取一个要访问的服务

2.2如何替换

2.1 修改cloud-consumer-order80

2.2新建package : com.atguigu.myrule,上面包下新建MySelfRule规则类

@Configuration
public class MySelfRule {

    @Bean
    public IRule myRule(){
        return new RandomRule();//定义为随机
    }
}
1
2
3
4
5
6
7
8

2.3主启动类添加@RibbonClient

@EnableEurekaClient
@SpringBootApplication
@RibbonClient(name = "CLOUD-PAYMENT-SERVICE",configuration = MySelfRule.class)
public class OrderMain80 {
    public static void main(String[] args) {
        SpringApplication.run(OrderMain80.class,args);
    }
}
1
2
3
4
5
6
7
8

2.4测试：

http://localhost/consumer/payment/get/3
发现port返回是随机的，不再是轮询。

3.Ribbon负载均衡算法

3.1 RoundRobinRule源码解析

核心算法：next = (current + 1) % lb.getAllServers().size();
返回下标=（当前下标+1）对节点的数量取余，最终返回的下标总是小于等于节点数-1；

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

            while(true) {
                if (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) {
                        int nextServerIndex = this.incrementAndGetModulo(serverCount);
                        server = (Server)allServers.get(nextServerIndex);
                        if (server == null) {
                            Thread.yield();
                        } else {
                            if (server.isAlive() && server.isReadyToServe()) {
                                return server;
                            }

                            server = null;
                        }
                        continue;
                    }

                    log.warn("No up servers available from load balancer: " + lb);
                    return null;
                }

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

                return server;
            }
        }
    }
// do while 循环至少执行一次循环体，当返回结果为true,则继续执行循环体，返回false,退出循环体
// AtomicInteger的compareAndSet(int expect, int update);使用了CAS算法，只有当expect是期待的值才会更新
    private int incrementAndGetModulo(int modulo) {
        int current;
        int next;
        do {
            current = this.nextServerCyclicCounter.get();
            next = (current + 1) % modulo;
        } while(!this.nextServerCyclicCounter.compareAndSet(current, next));
				// 返回false,退出循环体
        return next;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

3.2 手写轮询负载均衡

3.2.1 8001/8002微服务改造

@GetMapping(value = "/payment/lb")
public String getPaymentLB(){
    return serverPort;
}
1
2
3
4

3.2.2 80订单微服务改造

ApplicationContextBean去掉@LoadBalanced

新建LoadBalancer接口

public interface LoadBalancer {
     // 收集服务器总共有多少台能够提供服务的机器，并放到list里面
    ServiceInstance instances(List<ServiceInstance> serviceInstances);

1
2
3
4

新建LoadBalancerImpl实现类，标注@Component注解，位置位于启动类子包下

@Component
public class LoadBalancerImpl implements LoadBalancer {

    private AtomicInteger atomicInteger = new AtomicInteger(0);

    private final int getAndIncrement(int size){
        for (;;){
            int current = this.atomicInteger.get();
            int next = (current+1)%size;
            // 如果current的值没改变，就返回next
            if(this.atomicInteger.compareAndSet(current,next)){
            // 返回true,退出循环体
                return next;
            }
        }
    }

    @Override
    public ServiceInstance instances(List<ServiceInstance> serviceInstances) {
        // 得到服务器的下标位置
        int index = getAndIncrement(serviceInstances.size());
        return serviceInstances.get(index);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

修改OrderController

    @Resource
    private LoadBalancer loadBalancer;

    @Resource
    private DiscoveryClient discoveryClient;
    
@GetMapping(value = "/consumer/payment/lb")
     public String getPaymentLB(){
        List<ServiceInstance> instances = discoveryClient.getInstances("CLOUD-PAYMENT-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);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

测试：http://localhost/consumer/payment/lb
停掉8001服务，测试发现discoveryClient.getInstances返回的是在线的服务节点。如果关闭了eureka的自我保护，那么返回的port就一直是8002