之前分析了Feign组件，今天我们来研究 Netflix 团队开发的另外一个组件--Ribbon。
Ribbon 和 Feign 有很多相似的地方，首先，它们本质上都是 HTTP client，其次，它们都具备重试、集成断路器等功能。最大的区别在于，Ribbon 内置了一个负载均衡器，而 Feign 没有。

本文将介绍如何使用原生的 Ribbon，注意是原生的，而不是被 Spring 层层封装的 Ribbon。

为什么要使用Ribbon

这里我们需要回答两个问题：

1. 为什么要使用 HTTP client？
2. 为什么要在 HTTP client 里内置负载均衡器？

其中，第一个问题在Feign中已经讲过，这里就不啰嗦了，我们直接看第二个问题。

我们知道，Apache HTTP client、Feign 并没有内置负载均衡器，也就是说，HTTP client 并不一定要内置负载均衡器，那为什么 Ribbon 要搞特殊呢？

其实，我们可以想想，Ribbon 更多地被用在内部调用，而这种场景有一个比较大的特点--目标服务为集群部署。通常情况下，在调用目标服务时，我们希望请求尽可能平均地分发到每个实例。通过内置的负载均衡器，Ribbon 可以很好地满足要求，而 Apache HTTP client、Feign 就无法做到。

所以，在 HTTP client 里内置负载均衡器是为了能够在目标服务为集群部署时提供负载均衡支持。

 

有的人可能会说，你单独部署一台负载均衡器就行了嘛，搞那么复杂干嘛。当然，你可以这么做。但是你要考虑很重要的一点，mid-tier services 的请求量要远大于 edge services，所以你需要一台性能极高的负载均衡器。从这个角度来说，Ribbon 的方案帮你省下了独立部署负载均衡器的开销。

 

如何使用Ribbon

项目中我用 RxNettty 写了一个简单的 HTTP 接口（见cn.zzs.ribbon.RxUserServer）供后面的例子调用，这个接口运行在本机的 8080、8081、8082 接口，用来模拟三台不同的实例。所以，如果你想要测试项目中的例子，要先把这三台实例先启动好。

http://127.0.0.1:8080/user/getUserById?userId={userId}
request:userId=1
response:User [id=1, name=zzs001, age=18]

这里提醒一下，Ribbon 的 API 用到了很多 RxJava 代码，如果之前没接触过，最好先了解下。

作为HTTP client的用法

和 Feign 一样，Ribbon 支持使用注解方式定义 HTTP 接口，除此之外，Ribbon 还支持使用HttpRequestTemplate、HttpClientRequest等方式定义，这部分的例子我也提供了，感兴趣可以移步项目源码。

服务实例的列表通过ConfigurationManager设置。当你看到ConfigurationManager时，会不会觉得很熟悉呢？我们之前Eureka配置体系中详细介绍过，没错，Ribbon 用的还是这套配置体系。需要强调下，Netflix 团队开发的这套配置体系提供了动态配置支持（当然，你要会用才行），就拿本项目为例，如果运行过程中你通过ConfigurationManager更改了 listOfServers，那么，Ribbon 会感知到这种变化，使用最新的 listOfServers 来负载均衡，只是会有一点点延迟。

// 使用注解定义HTTP API
@ClientProperties(properties = {
        @Property(name="ReadTimeout", value="2000"),
        @Property(name="ConnectTimeout", value="1000"),
        @Property(name="MaxAutoRetries", value="1"),
        @Property(name="MaxAutoRetriesNextServer", value="2")
}, exportToArchaius = true)
interface UserService {
    @TemplateName("getUserById")
    @Http(
            method = HttpMethod.GET,
            uri = "/user/getUserById?userId={userId}",
            headers = {
                    @Header(name = "X-Platform-Version", value = "xyz"),
                    @Header(name = "X-Auth-Token", value = "abc")
            })
    RibbonRequest<ByteBuf> getUserById(@Var("userId") String userId);
}

public class RxUserProxyTest {
    @Test
    public void testBase() throws InterruptedException {
        // 指定服务实例的地址
        // key：服务+“.ribbon.”+配置项名称（见com.netflix.client.config.CommonClientConfigKey）
        ConfigurationManager.getConfigInstance().setProperty(
                "UserService.ribbon.listOfServers", "127.0.0.1:8080,127.0.0.1:8081,127.0.0.1:8082");
        
        UserService userService = Ribbon.from(UserService.class);
        
        userService.getUserById("1")
            .toObservable()
            .subscribe(new Subscriber<Object>() {
                @Override
                public void onCompleted() {
                    LOG.info("onCompleted");
                }
                
                @Override
                public void one rror(Throwable e) {
                    e.printStackTrace();
                }
                
                @Override
                public void onNext(Object t) {
                    LOG.info("onNext:{}", t);
                    if(t != null && t instanceof ByteBuf) {
                        LOG.info(ByteBuf.class.cast(t).toString(Charset.defaultCharset()));
                    }
                }
            });
        // 因为请求HTTP接口是异步的，这里要让测试主线程先睡一会
        Thread.sleep(10000);
    }
}

默认的负载均衡规则

为了观察多次请求在三台实例的分配情况，现在我们更改下代码，试着发起 6 次请求。

    @Test
    public void test01() throws InterruptedException {
        ConfigurationManager.getConfigInstance().setProperty(
                "UserService.ribbon.listOfServers", "127.0.0.1:8080,127.0.0.1:8081,127.0.0.1:8082");
            
        UserService userService = Ribbon.from(UserService.class);
        // 发起多次请求
        Observable<ByteBuf>[] requestList = new Observable[]{
                userService.getUserById("1").toObservable(),
                userService.getUserById("2").toObservable(),
                userService.getUserById("3").toObservable(),
                userService.getUserById("4").toObservable(),
                userService.getUserById("5").toObservable(),
                userService.getUserById("6").toObservable()
        };
        Observable.concat(Observable.from(requestList))
                .subscribe(subscriber);
        Thread.sleep(10000);
    }

运行测试，可以看到，6 次请求被平均地分配到了 3 台实例。

在日志中，可以看到了默认的负载均衡规则。

通过源码可以看到，这个默认的规则本质上采用的是轮询策略RoundRobinRule。除此之外，Ribbon 还定义了RandomRule、RetryRule等规则供我们选择。

public class AvailabilityFilteringRule {
    RoundRobinRule roundRobinRule = new RoundRobinRule();
}

自定义负载均衡规则

自定义负载均衡规则需要继承com.netflix.loadbalancer.AbstractLoadBalancerRule，并实现 choose 方法。这里我定义的规则是：不管有多少实例，默认访问第一台。

public class MyLoadBalancerRule extends AbstractLoadBalancerRule {
    @Override
    public Server choose(Object key) {
        
        ILoadBalancer lb = getLoadBalancer();
        
        List<Server> allServers = lb.getAllServers();
        
        return allServers.stream().findFirst().orElse(null);
    }
}

接着，只需要通过ConfigurationManager配置自定义规则就行。

    @Test
    public void test01() throws InterruptedException {
        ConfigurationManager.getConfigInstance().setProperty(
                "UserService.ribbon.listOfServers", "127.0.0.1:8080,127.0.0.1:8081,127.0.0.1:8082");
        // 配置自定义规则
        ConfigurationManager.getConfigInstance().setProperty(
                "UserService.ribbon.NFLoadBalancerRuleClassName", "cn.zzs.ribbon.MyLoadBalancerRule");
            
        UserService userService = Ribbon.from(UserService.class);
        
        Observable<ByteBuf>[] requestList = new Observable[]{
                userService.getUserById("1").toObservable(),
                userService.getUserById("2").toObservable(),
                userService.getUserById("3").toObservable(),
                userService.getUserById("1").toObservable(),
                userService.getUserById("2").toObservable(),
                userService.getUserById("3").toObservable()
        };
        Observable.concat(Observable.from(requestList))
                .subscribe(subscriber);
        Thread.sleep(10000);
    }

运行测试，可以看到，所有请求都被分配到了第一台实例。自定义负载均衡规则生效。

 

动态刷新listOfServers

在前面的例子中，我们的服务实例地址 listOfServers 都是写死的，然而，在实际项目中，目标服务的实例数量、地址都是变化的，所以，我们需要动态地更新 listOfServers，而不能写死。

上面说过 Netflix 开发的这套配置体系是支持动态配置的，所以，我能想到的最简方案就是，开一个任务，定时地从注册中心等地方拉取最新的实例列表，再把这个新的列表塞进ConfigurationManager就行了。

针对以 eureka server 为注册中心的项目，官方提供了一个实现方案。对比我上面说的方案，这个方案逻辑虽简单，但实现更复杂，源码可读性挺高的，感兴趣的朋友可以看看。

    @Test
    public void testEureka() throws InterruptedException {
        // 指定实例列表从eureka获取
        ConfigurationManager.getConfigInstance().setProperty(
                "UserService.ribbon.NIWSServerListClassName", "com.netflix.niws.loadbalancer.DiscoveryEnabledNIWSServerList");
        ConfigurationManager.getConfigInstance().setProperty(
                "UserService.ribbon.DeploymentContextBasedVipAddresses", "UserService");
        
        // 初始化EurekaClient
        DiscoveryManager.getInstance().initComponent(new MyDataCenterInstanceConfig(), new DefaultEurekaClientConfig());
        
        UserService userService = Ribbon.from(UserService.class);
        userService.getUserById("1")
            .toObservable()
            .subscribe(subscriber);
        
        Thread.sleep(10000);
    }
}

结语

以上，基本讲完 Ribbon 的使用方法，其实 Ribbon 还有其他可以扩展的东西，例如，断路器、重试等等。感兴趣的话，可以自行分析。

标签：分析,toObservable,负载,getUserById,userService,组件,UserService,Ribbon
来源： https://www.cnblogs.com/johnvwan/p/15575913.html