Ribbon远程调用

摘要:原创地址:https://zijiancode.cn/archives/ribbon 欢迎转载,转载时请保留摘要,谢谢!

文章中的样例会使用Nacos篇中的服务,读者可以看文章也可以直接结合gitee的代码观看

gitee: https://gitee.com/lzj960515/my-micro-service-demo.git

什么是Ribbon

Ribbon是由Netflix公司开发的,一个客户端的IPC(进程间通信)库,它提供了以下特性

  • 负载均衡
  • 容错
  • 支持多协议(HTTP、TCP、UDP)
  • 缓存和批处理

github地址:https://github.com/Netflix/ribbon/

什么是客户端的负载均衡?

在进程间通信时,服务提供者(服务端)具备多个实例,由服务消费者(客户端)通过负载均衡算法自主选择某个实例发起调用,就是客户端的负载均衡。

Ribbon远程调用

与之对应的便是服务端的负载均衡,常见如Nginx

Ribbon入门

1.引入Ribbon相关依赖

<dependencies>
  <dependency>
    <groupId>com.netflix.ribbon</groupId>
    <artifactId>ribbon</artifactId>
    <version>2.3.0</version>
  </dependency>
  <dependency>
    <groupId>com.netflix.ribbon</groupId>
    <artifactId>ribbon-core</artifactId>
  </dependency>
  <dependency>
    <groupId>com.netflix.ribbon</groupId>
    <artifactId>ribbon-loadbalancer</artifactId>
  </dependency>
  <dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
  </dependency>
  <dependency>
    <groupId>io.reactivex</groupId>
    <artifactId>rxjava</artifactId>
  </dependency>
  <dependency>
    <groupId>io.reactivex</groupId>
    <artifactId>rxnetty</artifactId>
    <version>0.4.9</version>
  </dependency>
  <dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
  </dependency>
  <!-- 用于发送请求,当然也可以有JDK自带的 -->
  <dependency>
    <groupId>cn.hutool</groupId>
    <artifactId>hutool-http</artifactId>
    <version>5.6.3</version>
  </dependency>
  <dependency>
    <groupId>commons-logging</groupId>
    <artifactId>commons-logging</artifactId>
    <version>1.2</version>
  </dependency>
</dependencies>

2. 编写代码

public class RibbonDemo {

    public static void main(String[] args) throws Exception {
        // 重试handler 第一个参数为调用同一个实例的次数,第二个为再次尝试调用其他实例的个数,1就是调用失败后,换一个实例再来一次,还是失败就返回失败。
        final RetryHandler retryHandler = new DefaultLoadBalancerRetryHandler(0, 1, true);
        List<Server> serverList = Lists.newArrayList(
                new Server("127.0.0.1", 8083),
                new Server("127.0.0.1", 8084));
        // 创建一个负载均衡器,默认负载均衡算法为轮询
        ILoadBalancer loadBalancer = LoadBalancerBuilder.newBuilder()
                .buildFixedServerListLoadBalancer(serverList);
        for (int i = 0; i < 6; i++) {
            String result = LoadBalancerCommand.<String>builder()
                    .withLoadBalancer(loadBalancer)
                    .withRetryHandler(retryHandler)
                    .build()
                    .submit(server -> {
                        String url = "http://" + server.getHost() + ":" + server.getPort() + "/integral/remain";
                        return Observable.just(HttpUtil.get(url));
                    }).toBlocking().first();
            System.out.println(result);
        }
    }
}

这里我使用了Nacos篇的积分服务,小伙伴可以自己随便起一个服务。

3. 测试

起两个实例,端口分别为8083和8084,测试结果:

您当前的积分为:31 服务端口:8084
您当前的积分为:78 服务端口:8083
您当前的积分为:37 服务端口:8084
您当前的积分为:31 服务端口:8083
您当前的积分为:59 服务端口:8084
您当前的积分为:54 服务端口:8083

停掉8083,再次测试

您当前的积分为:66 服务端口:8084
您当前的积分为:32 服务端口:8084
您当前的积分为:52 服务端口:8084
您当前的积分为:87 服务端口:8084
您当前的积分为:66 服务端口:8084
您当前的积分为:46 服务端口:8084

可以发现6次都是成功,并且都调到了8084端口的服务

将RetryHandler改为以下配置后测试

RetryHandler retryHandler = new DefaultLoadBalancerRetryHandler(0, 0, true);
您当前的积分为:59 服务端口:8084
Exception in thread "main" cn.hutool.core.io.IORuntimeException: ConnectException: Connection refused (Connection refused)

第一次调用了8084端口的服务,第二次调用8083端口的服务时抛出了异常

Spring Cloud 整合Ribbon

了解了Ribbon的基本使用后,接下来就是学习如何整合到微服务系统中了,Netflix公司同样提供了一个spring-cloud-starter-netflix-ribbon供我们快速整合到Spring Cloud中。

github地址:https://github.com/spring-cloud/spring-cloud-netflix

注意:选择2.x的tag,3.x的分支已移除Ribbon

基本使用

1.在my-order服务中引入依赖

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
</dependency>

其实引入Nacos依赖时,Nacos已经引入了该依赖,这里就做做样子嘻嘻

2.配置RestTemplate

@LoadBalanced
@Bean
public RestTemplate restTemplate(){
  return new RestTemplate();
}

3.发起调用

restTemplate.getForObject("http://my-goods/goods/get", String.class);

这就没了?嗯,这就没了~

RestTemplate Interceptor

看完基本使用后,小伙伴肯定觉得很神奇,就加了个@LoadBalanced注解,就大功告成了?

我们现在就来简单唠一下他的原理吧。

RestTemplate中,实现了拦截器机制,举个栗子

public class RestTemplateDemo {

    public static void main(String[] args) {
        RestTemplate restTemplate = new RestTemplate();
        List<ClientHttpRequestInterceptor> interceptors = new ArrayList<>();
        interceptors.add(new RestTemplateInterceptor());
        restTemplate.setInterceptors(interceptors);
        System.out.println(restTemplate.getForObject("http://127.0.0.1:8083/integral/remain", String.class));
    }

    static class RestTemplateInterceptor implements ClientHttpRequestInterceptor {

        @Override
        public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) throws IOException {
            URI uri = request.getURI();
            System.out.println(uri.getRawPath());
            return execution.execute(request, body);
        }
    }
}

发起调用前添加了自定义拦截器RestTemplateInterceptor,调用时则将进入到intercept方法

@LoadBalanced注解便是在服务启动时往RestTemplate中添加了一个拦截器LoadBalancerInterceptor

测试效果:

12:43:33.585 [main] DEBUG org.springframework.web.client.RestTemplate - HTTP GET http://127.0.0.1:8083/integral/remain
12:43:33.603 [main] DEBUG org.springframework.web.client.RestTemplate - Accept=[text/plain, application/json, application/*+json, */*]
/integral/remain
12:43:33.773 [main] DEBUG org.springframework.web.client.RestTemplate - Response 200 OK
12:43:33.775 [main] DEBUG org.springframework.web.client.RestTemplate - Reading to [java.lang.String] as "text/plain;charset=UTF-8"
您当前的积分为:83 服务端口:8083

现在我们来模拟一下Ribbon中将server替换成真实的ip地址

public class RestTemplateDemo {

    public static void main(String[] args) {
        RestTemplate restTemplate = new RestTemplate();
        List<ClientHttpRequestInterceptor> interceptors = new ArrayList<>();
        interceptors.add(new RestTemplateInterceptor());
        restTemplate.setInterceptors(interceptors);
        System.out.println(restTemplate.getForObject("http://my-goods/goods/get", String.class));
    }

    static class RestTemplateInterceptor implements ClientHttpRequestInterceptor {

        @SneakyThrows
        @Override
        public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) throws IOException {
            return execution.execute(new MyRequestWrapper(request), body);
        }
    }

    static class MyRequestWrapper extends HttpRequestWrapper {
        /**
         * 模拟注册中心存储服务信息
         */
        private final Map<String, String> serverMap = Maps.newHashMap("my-goods", "127.0.0.1:8081");


        public MyRequestWrapper(HttpRequest request) {
            super(request);
        }

        @SneakyThrows
        @Override
        public URI getURI() {
            URI uri = super.getRequest().getURI();
            String server = uri.getHost();
            // 模拟从注册中心取出真实ip
            String host = serverMap.get(server);
            // 替换URI
            return new URI(uri.getScheme() + "://" + host + uri.getRawPath());
        }
    }
}

这里自定义了一个MyRequestWrapper,并且重写了getURI方法,由该方法进行uri替换,模拟实现了Ribbon的将服务名替换真实ip的过程。

从serverMap里get服务信息的逻辑可以再进一步,封装成取出多个服务信息,使用负载均衡策略取出其中一个。

负载均衡策略

Ribbon中实现了许多的负载均衡策略,它们都实现于一个IRule接口

Ribbon远程调用

RetryRule: 重试策略,内部包含一个subRule,使用subRule(默认为轮询)选择服务,当服务不可用时,在配置的时间内一直重试,直接找到可用的服务或者超时。

RoundRobinRule: 轮询策略。

WeightedResponseTimeRule: 根据响应时间加权,响应时间越长,权重越小,被选中的可能性越低。

ZoneAvoidanceRule: 默认的负载均衡策略,根据服务的区域和可用性得到一个服务列表,然后轮询。没有区域的话等同于轮询。

AvailabilityFilteringRule: 使用轮询选择服务,对选择的服务进行状态判断,过滤掉一直连接失败的服务,直到找到正常可用的服务。

BestAvaliableRule: 选择并发请求最小的服务。

RandomRule: 随机策略。

NacosRule: Nacos提供的策略,同集群优先策略。

使用方式

使用@Bean注解

@Configuration
public class RuleConfiguration {

    @Bean
    public IRule rule(){
        return new NacosRule();
    }
}

以上方式将在全局生效

如果想要为每个服务设置不同的负载策略可按如下配置:

配置方式

my-goods:
	ribbon:
  	NFLoadBalancerRuleClassName: com.alibaba.cloud.nacos.ribbon.NacosRule

注意,此方式必须带有服务名,单独使用ribbon.NFLoadBalancerRuleClassName是不会生效的

自定义负载均衡策略

除了使用框架中自带的负载均衡策略,我们还可以实现自己的策略,比如相同版本优先调用

@Slf4j
public class VersionRule extends AbstractLoadBalancerRule {

    @Autowired
    private NacosDiscoveryProperties nacosDiscoveryProperties;

    @Autowired
    private NacosServiceManager nacosServiceManager;

    @Override
    public Server choose(Object key) {
        try {
            // 得到元数据信息
            Map<String, String> metadata = this.nacosDiscoveryProperties.getMetadata();
            // 取出版本信息
            String version = metadata.get("version");

            // 获取调用的服务列表
            String group = this.nacosDiscoveryProperties.getGroup();
            DynamicServerListLoadBalancer loadBalancer = (DynamicServerListLoadBalancer) getLoadBalancer();
            String name = loadBalancer.getName();

            NamingService namingService = nacosServiceManager
                    .getNamingService(nacosDiscoveryProperties.getNacosProperties());
            List<Instance> instances = namingService.selectInstances(name, group, true);
            if (CollectionUtils.isEmpty(instances)) {
                log.warn("no instance in service {}", name);
                return null;
            }

            List<Instance> instancesToChoose = instances;
            if (StringUtils.isNotBlank(version)) {
                // 筛选出相同版本的服务
                List<Instance> sameClusterInstances = instances.stream()
                        .filter(instance -> Objects.equals(version,
                                instance.getMetadata().get("version")))
                        .collect(Collectors.toList());
                if (!CollectionUtils.isEmpty(sameClusterInstances)) {
                    instancesToChoose = sameClusterInstances;
                }
                else {
                    log.warn(
                            "A cross-cluster call occurs,name = {}, version = {}, instance = {}",
                            name, version, instances);
                }
            }

            Instance instance = ExtendBalancer.getHostByRandomWeight2(instancesToChoose);

            return new NacosServer(instance);
        }
        catch (Exception e) {
            log.warn("NacosRule error", e);
            return null;
        }
    }

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

增加version配置:

spring:
  application:
    name: my-order
  main:
    allow-bean-definition-overriding: true
  cloud:
    nacos:
      discovery:
        server-addr: 192.168.1.11:8850
        namespace: public
        username: nacos
        password: nacos
        metadata:
          version: 1.0

metadata是个map结构,键值对可自定义

服务容错

有时候请求一个服务出现网络抖动或者其他问题时,导致请求失败,此时我们希望可以重试或者换一个服务发起调用,就可以加上如下配置

ribbon:
  # 同服务重试次数
  MaxAutoRetries: 1
  # 再次尝试调用其他实例的个数
  MaxAutoRetriesNextServer: 1
  # 所以操作都发起重试,默认只重试get请求
  OkToRetryOnAllOperations: true
  # 服务连接超时时间
  ConnectTimeout: 1000
  # 服务响应超时时间
  ReadTimeout: 2000
  # 使用restclient 否则以上配置无效
  restclient:
    enabled: true

完整配置

ribbon:
  # 饥饿加载,启动时就创建客户端
  eager-load:
    enabled: true
    clients:
      - my-goods
  # 同服务重试次数
  MaxAutoRetries: 1
  # 再次尝试调用其他实例的个数
  MaxAutoRetriesNextServer: 1
  # 所以操作都发起重试,默认只重试get请求
  OkToRetryOnAllOperations: false
  # 服务连接超时时间
  ConnectTimeout: 1000
  # 服务响应超时时间
  ReadTimeout: 2000
  restclient:
    enabled: true

my-goods:
  ribbon:
    NFLoadBalancerRuleClassName: com.alibaba.cloud.nacos.ribbon.NacosRule

关于Ribbon的配置信息可查看类:CommonClientConfigKey

小结

本篇介绍了Ribbon相关知识,什么是Ribbon,如何使用,以及如何集成到Spring Cloud中,最后还介绍了如何自定义负载均衡策略。

看完之后想必有所收获吧~ 想要了解更多精彩内容,欢迎关注公众号:程序员阿鉴,阿鉴在公众号欢迎你的到来~

上一篇:二刷Ribbon源码


下一篇:Spring Cloud微服务组件Ribbon之负载均衡源码分析