7. Ribbon负载均衡调用

7.1 介绍

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

简单的说，Ribbon是Netflix发布的开源项目，主要功能是提供客户端的软件负载均衡算法和服务调用。Ribbon客户端组件提供一系列完善的配置项如连接超时，重试等。简单的说，就是在配置文件中列出Load Balancer（简称LB）后面所有的机器，Ribbon会自动的帮助你基于某种规则（如简单轮询，随机连接等）去连接这些机器。我们很容易使用Ribbon实现自定义的负载均衡算法。

7.2 RestTemplate+Ribbon实现负载均衡

7.3 核心组件IRule

7.3.1 核心组件自带的算法

7.3.2 如何替换默认规则

由于Ribbon是客户端负载算法，所以需要修改微服务客户端（远程调用发起方）。

• 在客户端定义Ribbon的配置类，用于在上下文中注册规则类。(根据官网说明,自定义的类最好在@ComponentScan注解扫描不到的包，这样的方便个性化配置，否则会导致调用所有服务都是用同一个规则) 自定义配置类如下：

@Configuration
public class RibbonConfig {
    @Bean
    public IRule ribbonRule(){
        System.out.println("准备负载规则....");
        return new RandomRule();
    }
}

• 在主启动类上使用@RibbonClient注解用于定义某一个服务所使用的的配置

@RibbonClient(name = "payment",configuration = RibbonConfig.class)

其中name为调用服务的名称，configuration为上述配置类

7.3.3 RoundRobinRule轮训算法源码分析

/*
 *
 * Copyright 2013 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */
package com.netflix.loadbalancer;

import com.netflix.client.config.IClientConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

/**
 * The most well known and basic load balancing strategy, i.e. Round Robin Rule.
 *
 * @author stonse
 * @author Nikos Michalakis <nikos@netflix.com>
 *
 */
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 7 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) {
    }
}

上述代码中 incrementAndGetModulo 方法涉及到自旋锁，请参考 博客

7.3.4 自定义算法

当有需要时可以自定义负载均衡算法，仿照RoundRobinRule定义算法，每5秒切换一次微服务

package org.jshand.ribbon;

import com.netflix.client.config.IClientConfig;
import com.netflix.loadbalancer.AbstractLoadBalancerRule;
import com.netflix.loadbalancer.ILoadBalancer;
import com.netflix.loadbalancer.Server;
import lombok.extern.slf4j.Slf4j;

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

@Slf4j
public class MySelfRule extends AbstractLoadBalancerRule {

    private AtomicInteger nextServerCyclicCounter;
    long lastTime = 0;
    int second_interval = 5;
    int index = 0;

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

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

    @Override
    public Server choose(Object key) {
        ILoadBalancer lb = getLoadBalancer();
        List<Server> servers = lb.getReachableServers();

        //计算与上次时间的间隔描述
        long time = (System.currentTimeMillis() - lastTime) / 1000 ;

        //与上次间隔超过5秒则切换 index
        if (time >=second_interval) {
            index = incrementAndGetModulo(servers.size());
            lastTime = System.currentTimeMillis();
        }
        log.info(System.currentTimeMillis() +"\t"+lastTime+"\t"+time);

        return servers.get(index);//根据index返回数据
    }


    /**
     * 使用原子性int(AtomicInteger) 在总数中递增
     * @param serverCount
     * @return
     */
    private int incrementAndGetModulo(int serverCount) {
        for (; ; ) {
            int current = nextServerCyclicCounter.get();
            int next = (current + 1) % serverCount;
            if (nextServerCyclicCounter.compareAndSet(current, next))
                return next;
        }
    }

}