springboot+zuul（二）------智能負載

一、參考

參考資料：https://www.cnblogs.com/flying607/p/8330551.html

ribbon+spring retry重試策略源碼分析：https://blog.csdn.net/xiao_jun_0820/article/details/79320352

 

二、背景

這幾天在做服務的高可用。

為了確保提供服務的某一台機器出現故障導致客戶的請求不可用，我們需要對這台服務器做故障重試或者智能路由到下一個可用服務器。

為此，特地上網查了些資料，最后選用了ribbon+spring retry的重試策略。

 

從參考的技術文章中可以看出，故障重試的核心

1是引入spring retry的依賴

        <dependency>
            <groupId>org.springframework.retry</groupId>
            <artifactId>spring-retry</artifactId>
            <version>1.2.2.RELEASE</version>
        </dependency>

 

2是開啟zuul和ribbon重試配置

zuul:
  retryable: true    #重試必配
ribbon:
  MaxAutoRetriesNextServer: 2     #更換服務實例次數
  MaxAutoRetries: 0        #當前服務重試次數
  OkToRetryOnAllOperations: true   #設置成false,只處理get請求故障

 

當然本文的目的並不止於此。

添加完了這些配置后，我們發現依然存在一些局限性。

1、當提供服務的集群機器實例小於MaxAutoRetriesNextServer時，只有采用輪詢策略的負載可以正常使用。

2、當提供服務的集群機器實例大於MaxAutoRetriesNextServer時，采取輪詢或者隨機策略的負載偶爾可以正常使用。

而采用最小並發策略，或者單一負載（一般是為了解決session丟失問題，即同一個客戶端發的請求固定訪問某個服務器）則

完全不能正常工作。

      為什么這么說呢？比如我們有5台機器提供服務，第一台機器可以正常提供服務，第二台並發量最小。

      當第二到第五台服務器掛掉以后，采用輪詢方式且MaxAutoRetriesNextServer=2。那么，ribbon會嘗試訪問第三台、第四台服務器。

      結果不言而喻。當然如果運氣好，第三台或第四台服務器是可以用的，那就能正常提供服務。

      采用隨機策略，同樣要依靠運氣。

      最小並發或單一策略的，則是不論重試幾次則因為總是選擇掛掉的第二個節點而完全失效。
那么，有什么解決辦法呢？

 

三、動態設置MaxAutoRetriesNextServer

出現這些問題，一個關鍵是MaxAutoRetriesNextServer被寫死了，而我們的提供server的數量又可能隨着集群的負載情況增加（減少並不影響）。

總不能因為每次增加服務器數量就改一次MaxAutoRetriesNextServer配置吧？既然不想改配置，那當然就是動態設置MaxAutoRetriesNextServer的值啊。

 

翻看重試的源碼 RibbonLoadBalancedRetryPolicy.java

    @Override
    public boolean canRetryNextServer(LoadBalancedRetryContext context) {
        //this will be called after a failure occurs and we increment the counter
        //so we check that the count is less than or equals to too make sure
        //we try the next server the right number of times
        return nextServerCount <= lbContext.getRetryHandler().getMaxRetriesOnNextServer() && canRetry(context);
    }

 

可以看出MaxAutoRetriesNextServer的值是從DefaultLoadBalancerRetryHandler里面獲取的。但是DefaultLoadBalancerRetryHandler又不提供設置MaxAutoRetriesNextServer的接口。

往上追溯DefaultLoadBalancerRetryHandler實例化的源碼

    @Bean
    @ConditionalOnMissingBean
    public RibbonLoadBalancerContext ribbonLoadBalancerContext(ILoadBalancer loadBalancer,
            IClientConfig config, RetryHandler retryHandler) {
        return new RibbonLoadBalancerContext(loadBalancer, config, retryHandler);
    }

    @Bean
    @ConditionalOnMissingBean
    public RetryHandler retryHandler(IClientConfig config) {
        return new DefaultLoadBalancerRetryHandler(config);
    }

發現DefaultLoadBalancerRetryHandler對象可以從RibbonLoadBalancerContext實例中獲取， 而RibbonLoadBalancerContext卻可以從SpringClientFactory獲取，那么我們只要新建retryHandler並重新賦值給RibbonLoadBalancerContext就可以了。

 

代碼：

1、將IClientConfig托管到spring上

    @Bean
    public IClientConfig ribbonClientConfig() {
        DefaultClientConfigImpl config = new DefaultClientConfigImpl();
        config.loadProperties(this.name);
        config.set(CommonClientConfigKey.ConnectTimeout, DEFAULT_CONNECT_TIMEOUT);
        config.set(CommonClientConfigKey.ReadTimeout, DEFAULT_READ_TIMEOUT);
        return config;
    }

 

2、新建retryHandler並更新到RibbonLoadBalancerContext

private void setMaxAutoRetiresNextServer(int size) {　　//size: 提供服務的集群數量
        SpringClientFactory factory = SpringContext.getBean(SpringClientFactory.class); //獲取spring托管的單例對象
        IClientConfig clientConfig = SpringContext.getBean(IClientConfig.class);
        int retrySameServer = clientConfig.get(CommonClientConfigKey.MaxAutoRetries, 0);//獲取配置文件中的值， 默認0
        boolean retryEnable = clientConfig.get(CommonClientConfigKey.OkToRetryOnAllOperations, false);//默認false。
        RetryHandler retryHandler = new DefaultLoadBalancerRetryHandler(retrySameServer, size, retryEnable);//新建retryHandler
        factory.getLoadBalancerContext(name).setRetryHandler(retryHandler);
    }

MaxAutoRetriesNextServer動態設置的問題就解決了。

 

四、剔除不可用的服務。

Eureka好像有提供服務的剔除和恢復功能，所以如果有用Eureka注冊中心，就不用往下看了。具體配置我也不太清楚。

因為我們沒用到eureka，所以在故障重試的時候，獲取到的服務列表里依然包含了掛掉的服務器。

這樣會導致最小並發策略和單一策略的負載出現問題。

跟蹤源碼，我們發現服務器故障后會調用canRetryNextServer方法，那么不如就在這個方法里面做文章吧。

 

自定義RetryPolicy 繼承RibbonLoadBalancedRetryPolicy並且重寫canRetryNextServer

public class ServerRibbonLoadBalancedRetryPolicy extends RibbonLoadBalancedRetryPolicy {

    private RetryTrigger trigger;
    public ServerRibbonLoadBalancedRetryPolicy(String serviceId, RibbonLoadBalancerContext context, ServiceInstanceChooser loadBalanceChooser, IClientConfig clientConfig) {
        super(serviceId, context, loadBalanceChooser, clientConfig);
    }

    public void setTrigger(RetryTrigger trigger) {
        this.trigger = trigger;
    }

    @Override
    public boolean canRetryNextServer(LoadBalancedRetryContext context) {
        boolean retryEnable = super.canRetryNextServer(context);
        if (retryEnable && trigger != null) {
            //回調觸發
            trigger.exec(context);
        }
        return retryEnable;
    }

    @FunctionalInterface
    public interface RetryTrigger {
        void exec(LoadBalancedRetryContext context);
    }
}    

 

自定義RetryPolicyFactory繼承RibbonLoadBalancedRetryPolicyFactory並重寫create方法

public class ServerRibbonLoadBalancedRetryPolicyFactory extends RibbonLoadBalancedRetryPolicyFactory {
    private SpringClientFactory clientFactory;
    private ServerRibbonLoadBalancedRetryPolicy policy;
    private ServerRibbonLoadBalancedRetryPolicy.RetryTrigger trigger;

    public ServerRibbonLoadBalancedRetryPolicyFactory(SpringClientFactory clientFactory) {
        super(clientFactory);
        this.clientFactory = clientFactory;
    }

    @Override
    public LoadBalancedRetryPolicy create(String serviceId, ServiceInstanceChooser loadBalanceChooser) {
        RibbonLoadBalancerContext lbContext = this.clientFactory
                .getLoadBalancerContext(serviceId);
        policy = new ServerRibbonLoadBalancedRetryPolicy(serviceId, lbContext, loadBalanceChooser, clientFactory.getClientConfig(serviceId));
        policy.setTrigger(trigger);
        return policy;
    }

    public void setTrigger(ServerRibbonLoadBalancedRetryPolicy.RetryTrigger trigger) {
        policy.setTrigger(trigger);//跟上面是setTrigger不知道誰會先觸發，所以兩邊都設置了。
        this.trigger = trigger;
    }
}

 

把LoadBalancedRetryPolicyFactory托管到spring

    @Bean
    @ConditionalOnClass(name = "org.springframework.retry.support.RetryTemplate")
    public LoadBalancedRetryPolicyFactory loadBalancedRetryPolicyFactory(SpringClientFactory clientFactory) {
        return new ServerRibbonLoadBalancedRetryPolicyFactory(clientFactory);
    }

 

然后我們就可以在我們rule類上面實現RetryTrigger方法。

public class ServerLoadBalancerRule extends AbstractLoadBalancerRule implements ServerRibbonLoadBalancedRetryPolicy.RetryTrigger {

    private static final Logger LOGGER = LoggerFactory.getLogger(ServerLoadBalancerRule.class);
    /**
     * 不可用的服務器
     */
    private Map<String, List<String>> unreachableServer = new HashMap<>(256);
    /**
     * 上一次請求標記
     */
    private String lastRequest;

    @Autowired
    LoadBalancedRetryPolicyFactory policyFactory;

    @Override
    public Server choose(Object key) {
        //初始化重試觸發器
        retryTrigger();
        return getServer(getLoadBalancer(), key);
    }

    private Server getServer(ILoadBalancer loadBalancer, Object key) {
　　　　//從數據庫獲取服務列表
　　　　List<ServerAddress> addressList = getServerAddress();
　　　　setMaxAutoRetriesNextServer(addressList.size());

       //過濾不可用服務
    }

    private void retryTrigger() {
        RequestContext ctx = RequestContext.getCurrentContext();
        String batchNo = (String) ctx.get(Constant.REQUEST_BATCH_NO);
        if (!isLastRequest(batchNo)) {
            //不是同一次請求，清理所有緩存的不可用服務
            unreachableServer.clear();
        }

        if (policyFactory instanceof ServerRibbonLoadBalancedRetryPolicyFactory) {
            ((ServerRibbonLoadBalancedRetryPolicyFactory) policyFactory).setTrigger(this);
        }
    }

    private boolean isLastRequest(String batchNo) {
        return batchNo != null && batchNo.equals(lastRequest);
    }

    @Override
    public void exec(LoadBalancedRetryContext context) {
        RequestContext ctx = RequestContext.getCurrentContext();
　　　　　//UUID,故障重試不會發生變化。客戶每次請求時會產生新的batchNo,可以在preFilter中生成。　
        String batchNo = (String) ctx.get(Constant.REQUEST_BATCH_NO);
        lastRequest = batchNo;

        List<String> hostAndPorts = unreachableServer.get((String) ctx.get(Constant.REQUEST_BATCH_NO));
        if (hostAndPorts == null) {
            hostAndPorts = new ArrayList<>();
        }
        if (context != null && context.getServiceInstance() != null) {
            String host = context.getServiceInstance().getHost();
            int port = context.getServiceInstance().getPort();
            if (!hostAndPorts.contains(host + Constant.COLON + port))
                hostAndPorts.add(host + Constant.COLON + port);
            unreachableServer.put((String) ctx.get(Constant.REQUEST_BATCH_NO), hostAndPorts);
        }
    }
}

 

這樣，我們就拿到了不可用的服務了，然后在重試的時候過濾掉unreachableServer中的服務就可以了。

這里有一點要注意的是，MaxAutoRetriesNextServer的值必須是沒有過濾的服務列表的大小。

 

當然，有人會有疑問，如果服務器數量過多，重試時間超過ReadTimeout怎么辦？我這里也沒關於超時的設置，因為本身讓客戶等待過久就不是很合理的需求

所以配置文件里面設置一個合理的ReadTimeout就好了，在這個時間段里面如果重試沒取到可用的服務就直接拋超時的信息給客戶。

源碼地址： https://github.com/rxiu/study-on-road/tree/master/trickle-gateway