客户端负载均衡Ribbon

发表于 更新于 分类于 ， ， 阅读次数： Changyan： 本文字数： 2k 阅读时长 ≈ 7 分钟

引言

目前主流的负载方案分为两种：一种是集中式负载均衡，在消费者和服务提供方中间使用独立的代理方式进行负载，有硬件的（比如F5），也有软件的（比如 Nginx）。另一种则是客户端自己做负载均衡，根据自己的请求情况做负载， Ribbon就属于客户端自己量负载。如果用一句话介绍，那就是： Ribbon是 Netflix开源的一款用于客户端负载均衡的工具软件。

——《spring cloud 微服务 入门、进阶与实战》第52页

与spring cloud集成

在spring cloud 项目中集成Ribbon 只需要在pom.xml 中添加下面的依赖即可，如果用了Eureka也可以不用配置，因为Eureka 中已经引用了Ribbon

1
2
3
4
<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
</dependency>

RestTemplate负载均衡

新建一个spring cloud项目 命名为 springcloud-ribbon

1
2
3
#properties
spring.application.name=spring-cloud-ribbon
server.port=8083

新建一个配置类

1
2
3
4
5
6
7
8
@Configuration
public class BeanConfiguration {
    @Bean
    @LoadBalanced
    public RestTemplate getRestTemplate(){
        return new RestTemplate();
    }
}

编写controller 测试接口

1
2
3
4
5
6
7
8
9
10
11
@RestController
public class RibbonController {
    @Autowired
    private RestTemplate restTemplate;

    @GetMapping("/ribbon/hello")
    public String callHello(){
        return restTemplate.getForObject("http://eureka-client-service/user/hello",String.class);
    }

}

restTemplate 在这里通过服务实例的名字调用的时候，当注册在注册中心的多个服务都叫eureka-client-service名字的时候。就会自动做负载均衡（默认轮询策略）。

image-20200502112502789.png

@RestTemplate注解原理

相信大家一定有一个疑问：为什么在 Rest Template上加了一个@LoadBalanced之后Resttemplate就能够跟 Eureka结合了，不但可以使用服务名称去调用接口，还可以负载均衡应该归功于 Spring Cloud给我们做了大量的底层工作，因为它将这些都封装好了，我们用起来才会那么简单。框架就是为了简化代码，提高效率而产生的。

——《spring cloud 微服务 入门、进阶与实战》第58页

这里主要的逻辑就是给 Rest Template增加拦截器，在请求之前对请求的地址进行替换或者根据具体的负载策略选择服务地址，然后再去调用，这就是@Loadbalanced的原理。

自定义@MyLoadbalanced 来实现@Loadbalanced的功能：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

import org.springframework.beans.factory.annotation.Qualifier;

@Target({ ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD }) @Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Qualifier
public @interface MyLoadBalanced {

}

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
import java.io.IOException;
import java.net.URI;

import org.springframework.cloud.client.loadbalancer.LoadBalancerClient;
import org.springframework.cloud.client.loadbalancer.LoadBalancerRequestFactory;
import org.springframework.http.HttpRequest;
import org.springframework.http.client.ClientHttpRequestExecution;
import org.springframework.http.client.ClientHttpRequestInterceptor;
import org.springframework.http.client.ClientHttpResponse;
import org.springframework.util.Assert;
/**
 * @ClassName MyLoadBalancerInterceptor
 * @Author wuzhiyong
 * @Date 2020/5/2 11:58
 * @Version 1.0
 **/
public class MyLoadBalancerInterceptor implements ClientHttpRequestInterceptor {


    private LoadBalancerClient loadBalancer;
    private LoadBalancerRequestFactory requestFactory;

    public MyLoadBalancerInterceptor(LoadBalancerClient loadBalancer, LoadBalancerRequestFactory requestFactory) {
        this.loadBalancer = loadBalancer;
        this.requestFactory = requestFactory;
    }

    public MyLoadBalancerInterceptor(LoadBalancerClient loadBalancer) {
        this(loadBalancer, new LoadBalancerRequestFactory(loadBalancer));
    }

    @Override
    public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,
                                        final ClientHttpRequestExecution execution) throws IOException {
        final URI originalUri = request.getURI();
        String serviceName = originalUri.getHost();
        System.out.println("进入自定义的请求拦截器中" + serviceName);
        Assert.state(serviceName != null, "Request URI does not contain a valid hostname: " + originalUri);
        return this.loadBalancer.execute(serviceName, requestFactory.createRequest(request, body, execution));
    }

}
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
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.client.loadbalancer.LoadBalancerClient;
import org.springframework.cloud.client.loadbalancer.LoadBalancerRequestFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.client.ClientHttpRequestInterceptor;
import org.springframework.web.client.RestTemplate;

@Configuration
public class MyLoadBalancerAutoConfiguration {
    @MyLoadBalanced
    @Autowired(required = false)
    private List<RestTemplate> restTemplates = Collections.emptyList();

    @Bean
    public MyLoadBalancerInterceptor myLoadBalancerInterceptor(LoadBalancerClient loadBalancerClient,
                                                               LoadBalancerRequestFactory requestFactory) {
        return new MyLoadBalancerInterceptor(loadBalancerClient,requestFactory);
    }

    @Bean
    public SmartInitializingSingleton myLoadBalancedRestTemplateInitializer(LoadBalancerClient loadBalancerClient,
                                                                            LoadBalancerRequestFactory requestFactory) {
        return new SmartInitializingSingleton() {
            @Override
            public void afterSingletonsInstantiated() {
                for (RestTemplate restTemplate : MyLoadBalancerAutoConfiguration.this.restTemplates) {
                    List<ClientHttpRequestInterceptor> list = new ArrayList<>(restTemplate.getInterceptors());
                    list.add(myLoadBalancerInterceptor(loadBalancerClient,requestFactory));
                    restTemplate.setInterceptors(list);
                }
            }
        };
    }
}

替换上自己的 @MyLoadBalanced

1
2
3
4
5
6
    @Bean
//    @LoadBalanced
    @MyLoadBalanced
    public RestTemplate getRestTemplate(){
        return new RestTemplate();
    }

重启再访问测试接口地址 /ribbon/hello 控制台如下：

image-20200502124238291.png

具体逻辑可参看源代码：

1
2
3
4
5
6
7
8
// @Loadbalancer注解相关源代码
org.springframework.cloud.client.loadbalancer.LoadBalanced

org.springframework.cloud.client.loadbalancer.LoadBalancerEurekaAutoConfiguration

org.springframework.cloud.client.loadbalancer.LoadBalancerInterceptorConfig

org.springframework.cloud.client.loadbalancer.LoadBalancerInterceptor

Ribbon API 使用

当你有一些特殊的需求，想通过Ribbon获取对应的服务信息时。可以使用LoadBalancer Client来获取。下面代码来获取

1
2
3
4
5
6
7
8
@Autowired
private LoadBalancerClient loadBalancerClient;

@GetMapping("/choose")
public Object chooseUrl(){
    ServiceInstance instance = loadBalancerClient.choose("eureka-client-service");
    return instance;
}

访问后可得到类似如下内容：（监控端点没开启则会少很多）

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
54
55
56
57
58
59
60
61
{
  "serviceId": "eureka-client-service",
  "server": {
    "host": "localhost",
    "port": 8081,
    "id": "localhost:8081",
    "zone": "defaultZone",
    "readyToServe": true,
    "instanceInfo": {
      "instanceId": "eureka-client-service:localhost:8081",
      "app": "EUREKA-CLIENT-SERVICE",
      "ipAddr": "localhost",
      "sid": "na",
      "homePageUrl": "http://localhost:8081/",
      "statusPageUrl": "https://wu_zhiyong.gitee.io/myblog/",
      "healthCheckUrl": "http://localhost:8081/actuator/health",
      "vipAddress": "eureka-client-service",
      "secureVipAddress": "eureka-client-service",
      "countryId": 1,
      "dataCenterInfo": {
        "@class": "com.netflix.appinfo.InstanceInfo$DefaultDataCenterInfo",
        "name": "MyOwn"
      },
      "hostName": "localhost",
      "status": "UP",
      "overriddenStatus": "UNKNOWN",
      "leaseInfo": {
        "renewalIntervalInSecs": 5,
        "durationInSecs": 5,
        "registrationTimestamp": 1588387730005,
        "lastRenewalTimestamp": 1588400582649,
        "evictionTimestamp": 0,
        "serviceUpTimestamp": 1588387730005
      },
      "isCoordinatingDiscoveryServer": false,
      "metadata": {
        "create-user": "WuZhiYong",
        "management.port": "8081"
      },
      "lastUpdatedTimestamp": 1588387730005,
      "lastDirtyTimestamp": 1588387730000,
      "actionType": "ADDED"
    },
    "metaInfo": {
      "instanceId": "eureka-client-service:localhost:8081",
      "appName": "EUREKA-CLIENT-SERVICE",
      "serviceIdForDiscovery": "eureka-client-service"
    },
    "alive": true,
    "hostPort": "localhost:8081"
  },
  "secure": false,
  "metadata": {
    "create-user": "WuZhiYong",
    "management.port": "8081"
  },
  "host": "localhost",
  "port": 8081,
  "uri": "http://localhost:8081",
  "instanceId": "localhost:8081"
}

负载均衡策略介绍

负载策略代码整体如下，默认继承AbstractLoadBalancerRule

image-20200502143706413.png

image-20200502151313128.png

  • BestAvailabl：选择一个最小的并发请求的 Server，逐个考察 Server，如果 Server被标记为错误，则跳过，然后再选择 ActiveRequestCount 中最小的 Server

  • AvailabilityFilteringRule：过滤掉那些一直连接失败的且被标记为 circuit tripped的后端 Server.，并过滤掉那些高并发的后端 Server或者使用一个 Availability Predicate来包含过滤 Server的逻辑。其实就是检查 Status里记录的各个 Server E的运行状态。

  • ZoneAvoidancerule：使用 Zone CAvoidancepredicate 和 Availability Predicate来判断是否选择某个 Server，前一个判断判定一个zone的运行性能是否可用，别除不可用的Zone（的所有 Server）， Availabilitypredicate用于过滤掉连接数过多的 Servero.

  • RandomRule：随机选择一个 Server。

  • RoundRobinRule：轮询选择，轮询 index，选择 index对应位置的 Server 。

  • RetryRule：对选定的负载均衡策略机上重试机制，也就是说当选定了某个策略进行请求负载时在一个配置时间段内若选择 Server不成功，则一直尝试使用 subrule的方式选择一个可用的 Server。

  • ResponseTimeWeightRule：作用同 WeightResponseTimeRule。ResponseTimeWeightRule后来更名为WeightResponseTimeRule。

  • WeightResponseTimeRule根据响应时间分配一个 Weight（权重），响应时间越长，Weight 越小，被选中的可能性越低。

    ——《spring cloud 微服务 入门、进阶与实战》第65页

自定义负载策略

通过实现 IRule接口可以自定义负载策略，主要的选择服务逻辑在 choose方法中。这边只是演示怎么自定义负载策略，所以没写选择的逻辑，直接返回服务列表中第一个服务。

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
import java.util.List;

import com.netflix.loadbalancer.ILoadBalancer;
import com.netflix.loadbalancer.IRule;
import com.netflix.loadbalancer.Server;

public class MyRule implements IRule {

	private ILoadBalancer lb;

	@Override
	public Server choose(Object key) {
		List<Server> servers = lb.getAllServers();
		for (Server server : servers) {
			System.out.println(server.getHostPort());
		}
		return servers.get(0);
	}

	@Override
	public void setLoadBalancer(ILoadBalancer lb) {
		this.lb = lb;
	}

	@Override
	public ILoadBalancer getLoadBalancer() {
		return lb;
	}

}

配置自定义的负载策略。

  • 第一种方式

    把自定义策略注册到Bean

    1
    2
    3
    4
    5
    6
    7
    8
    @Configuration
    public class BeanConfiguration {

        @Bean
        public MyRule rule(){
            return new MyRule();
        }
    }

    给eureka-client-service(服务名) 关联配置

    1
    2
    3
    @RibbonClient(name = "eureka-client-service",configuration = BeanConfiguration.class)
    public class RibbonClientConfig {
    }

  • 第二种方式

    1
    2
    #给 eureka-client-service 服务配置自定义负载策略
    eureka-client-service.ribbon.NFLoadBalancerRuleClassName=com.study.ribbon.config.MyRule

支持配置的属性如下：

.ribbon.NFLoadBalancerClassName: 配置ILoadBalancer的实现类
.ribbon.NFLoadBalancerRuleClassName:配置IRule的实现类
.ribbon.NFLoadBalancerPingClassName: 配置IPing的实现类
.ribbon.NIWSServerListClassName: 配置ServerList的实现类
.ribbon.NIWSServerListFilterClassName:配置ServerListFilter的实现类

常用配置

  • 禁用Eureka

    1
    ribbon.eureka.enabled=false

    当禁用了Eureka 后就不能通过服务名称去调用接口了。必须指定调用地址

  • 配置接口地址列表

    1
    eureka-client-service.ribbon.listOfServers=localhost:8081,localhost:8082

    eureka-client-service是服务名称，这样配置后就可以通过服务名调用接口了。

  • 超时时间

    为每个服务配置超时时间

    1
    2
    3
    4
    #请求链接的超时时间
    eureka-client-service.ribbon.ConnectTimeout=2000
    #请求处理的超时时间
    eureka-client-service.ribbon.ReadTimeOut=5000

    为所有服务配置超时时间

    1
    2
    3
    4
    #请求链接的超时时间
    ribbon.ConnectTimeout=2000
    #请求处理的超时时间
    ribbon.ReadTimeOut=5000

  • 并发参数

    1
    2
    3
    4
    #最大链接数
    ribbon.maxToTalConnections=500
    #每个host最大链接数
    ribbon.MaxConnectionsPerHost=500

重试机制

除了使用 Ribbon 自带的重试策略，我们还可以通过集成 Spring Retry 来进行重试操作。

Spring Retry依赖

1
2
3
4
<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
</dependency>
1
2
3
4
5
6
7
8
# 对当前实例的重试次数
ribbon.maxAutoRetries=1
# 切换实例的重试次数
ribbon.maxAutoRetriesNextServer=3
# 对所有操作请求都进行重试
ribbon.okToRetryOnAllOperations=true
# 对Http响应码进行重试
ribbon.retryableStatusCodes=500,404,502

参考：

《spring cloud 微服务 入门、进阶与实战》

代码来源：

https://github.com/yinjihuan/spring-cloud/tree/master/Spring-Cloud-Book-Code-2/ch-4