专栏首页码匠的流水账聊聊ribbon的retry

聊聊ribbon的retry

本文主要研究一下ribbon的retry

配置

HttpClientRibbonConfiguration

spring-cloud-netflix-ribbon-2.0.0.RELEASE-sources.jar!/org/springframework/cloud/netflix/ribbon/apache/HttpClientRibbonConfiguration.java

    @Bean
    @ConditionalOnMissingBean(AbstractLoadBalancerAwareClient.class)
    @ConditionalOnClass(name = "org.springframework.retry.support.RetryTemplate")
    public RetryableRibbonLoadBalancingHttpClient retryableRibbonLoadBalancingHttpClient(
            IClientConfig config, ServerIntrospector serverIntrospector,
            ILoadBalancer loadBalancer, RetryHandler retryHandler,
            LoadBalancedRetryFactory loadBalancedRetryFactory, CloseableHttpClient httpClient) {
        RetryableRibbonLoadBalancingHttpClient client = new RetryableRibbonLoadBalancingHttpClient(
            httpClient, config, serverIntrospector, loadBalancedRetryFactory);
        client.setLoadBalancer(loadBalancer);
        client.setRetryHandler(retryHandler);
        Monitors.registerObject("Client_" + this.name, client);
        return client;
    }

OkHttpRibbonConfiguration

spring-cloud-netflix-ribbon-2.0.0.RELEASE-sources.jar!/org/springframework/cloud/netflix/ribbon/okhttp/OkHttpRibbonConfiguration.java

    @Bean
    @ConditionalOnMissingBean(AbstractLoadBalancerAwareClient.class)
    @ConditionalOnClass(name = "org.springframework.retry.support.RetryTemplate")
    public RetryableOkHttpLoadBalancingClient retryableOkHttpLoadBalancingClient(
        IClientConfig config,
        ServerIntrospector serverIntrospector,
        ILoadBalancer loadBalancer,
        RetryHandler retryHandler,
        LoadBalancedRetryFactory loadBalancedRetryFactory,
        OkHttpClient delegate) {
        RetryableOkHttpLoadBalancingClient client = new RetryableOkHttpLoadBalancingClient(delegate, config,
                serverIntrospector, loadBalancedRetryFactory);
        client.setLoadBalancer(loadBalancer);
        client.setRetryHandler(retryHandler);
        Monitors.registerObject("Client_" + this.name, client);
        return client;
    }

RetryableRibbonLoadBalancingHttpClient

spring-cloud-netflix-ribbon-2.0.0.RELEASE-sources.jar!/org/springframework/cloud/netflix/ribbon/apache/RetryableRibbonLoadBalancingHttpClient.java

/**
 * An Apache HTTP client which leverages Spring Retry to retry failed requests.
 * @author Ryan Baxter
 * @author Gang Li
 */
public class RetryableRibbonLoadBalancingHttpClient extends RibbonLoadBalancingHttpClient {
    private LoadBalancedRetryFactory loadBalancedRetryFactory;

    public RetryableRibbonLoadBalancingHttpClient(CloseableHttpClient delegate,
                                                  IClientConfig config, ServerIntrospector serverIntrospector,
                                                  LoadBalancedRetryFactory loadBalancedRetryFactory) {
        super(delegate, config, serverIntrospector);
        this.loadBalancedRetryFactory = loadBalancedRetryFactory;
    }

    @Override
    public RibbonApacheHttpResponse execute(final RibbonApacheHttpRequest request, final IClientConfig configOverride) throws Exception {
        final RequestConfig.Builder builder = RequestConfig.custom();
        IClientConfig config = configOverride != null ? configOverride : this.config;
        RibbonProperties ribbon = RibbonProperties.from(config);
        builder.setConnectTimeout(ribbon.connectTimeout(this.connectTimeout));
        builder.setSocketTimeout(ribbon.readTimeout(this.readTimeout));
        builder.setRedirectsEnabled(ribbon.isFollowRedirects(this.followRedirects));

        final RequestConfig requestConfig = builder.build();
        final LoadBalancedRetryPolicy retryPolicy = loadBalancedRetryFactory.createRetryPolicy(this.getClientName(), this);

        RetryCallback<RibbonApacheHttpResponse, Exception> retryCallback = context -> {
            //on retries the policy will choose the server and set it in the context
            //extract the server and update the request being made
            RibbonApacheHttpRequest newRequest = request;
            if (context instanceof LoadBalancedRetryContext) {
                ServiceInstance service = ((LoadBalancedRetryContext) context).getServiceInstance();
                validateServiceInstance(service);
                if (service != null) {
                    //Reconstruct the request URI using the host and port set in the retry context
                    newRequest = newRequest.withNewUri(UriComponentsBuilder.newInstance().host(service.getHost())
                            .scheme(service.getUri().getScheme()).userInfo(newRequest.getURI().getUserInfo())
                            .port(service.getPort()).path(newRequest.getURI().getPath())
                            .query(newRequest.getURI().getQuery()).fragment(newRequest.getURI().getFragment())
                            .build().encode().toUri());
                }
            }
            newRequest = getSecureRequest(newRequest, configOverride);
            HttpUriRequest httpUriRequest = newRequest.toRequest(requestConfig);
            final HttpResponse httpResponse = RetryableRibbonLoadBalancingHttpClient.this.delegate.execute(httpUriRequest);
            if (retryPolicy.retryableStatusCode(httpResponse.getStatusLine().getStatusCode())) {
                throw new HttpClientStatusCodeException(RetryableRibbonLoadBalancingHttpClient.this.clientName,
                        httpResponse, HttpClientUtils.createEntity(httpResponse), httpUriRequest.getURI());
            }
            return new RibbonApacheHttpResponse(httpResponse, httpUriRequest.getURI());
        };
        LoadBalancedRecoveryCallback<RibbonApacheHttpResponse, HttpResponse> recoveryCallback = new LoadBalancedRecoveryCallback<RibbonApacheHttpResponse, HttpResponse>() {
            @Override
            protected RibbonApacheHttpResponse createResponse(HttpResponse response, URI uri) {
                return new RibbonApacheHttpResponse(response, uri);
            }
         };
        return this.executeWithRetry(request, retryPolicy, retryCallback, recoveryCallback);
    }

    @Override
    public boolean isClientRetryable(ContextAwareRequest request) {
        return request!= null && isRequestRetryable(request);
    }

    private boolean isRequestRetryable(ContextAwareRequest request) {
        if (request.getContext() == null || request.getContext().getRetryable() == null) {
            return true;
        }
        return request.getContext().getRetryable();
    }

    private RibbonApacheHttpResponse executeWithRetry(RibbonApacheHttpRequest request, LoadBalancedRetryPolicy retryPolicy,
                                                      RetryCallback<RibbonApacheHttpResponse, Exception> callback,
                                                      RecoveryCallback<RibbonApacheHttpResponse> recoveryCallback) throws Exception {
        RetryTemplate retryTemplate = new RetryTemplate();
        boolean retryable = isRequestRetryable(request);
        retryTemplate.setRetryPolicy(retryPolicy == null || !retryable ? new NeverRetryPolicy()
                : new RetryPolicy(request, retryPolicy, this, this.getClientName()));
        BackOffPolicy backOffPolicy = loadBalancedRetryFactory.createBackOffPolicy(this.getClientName());
        retryTemplate.setBackOffPolicy(backOffPolicy == null ? new NoBackOffPolicy() : backOffPolicy);
        RetryListener[] retryListeners = this.loadBalancedRetryFactory.createRetryListeners(this.getClientName());
        if (retryListeners != null && retryListeners.length != 0) {
            retryTemplate.setListeners(retryListeners);
        }
        return retryTemplate.execute(callback, recoveryCallback);
    }

    @Override
    public RequestSpecificRetryHandler getRequestSpecificRetryHandler(RibbonApacheHttpRequest request, IClientConfig requestConfig) {
        return new RequestSpecificRetryHandler(false, false, RetryHandler.DEFAULT, null);
    }

    static class RetryPolicy extends InterceptorRetryPolicy {
        public RetryPolicy(HttpRequest request, LoadBalancedRetryPolicy policy,
                ServiceInstanceChooser serviceInstanceChooser, String serviceName) {
            super(request, policy, serviceInstanceChooser, serviceName);
        }
    }
}
  • 这里重点看executeWithRetry,通过设置RetryTemplate的RetryPolicy及BackOffPolicy,来执行重试策略

RetryableOkHttpLoadBalancingClient

spring-cloud-netflix-ribbon-2.0.0.RELEASE-sources.jar!/org/springframework/cloud/netflix/ribbon/okhttp/RetryableOkHttpLoadBalancingClient.java

/**
 * An OK HTTP client which leverages Spring Retry to retry failed request.
 * @author Ryan Baxter
 * @author Gang Li
 */
public class RetryableOkHttpLoadBalancingClient extends OkHttpLoadBalancingClient {

    private LoadBalancedRetryFactory loadBalancedRetryFactory;

    public RetryableOkHttpLoadBalancingClient(OkHttpClient delegate, IClientConfig config, ServerIntrospector serverIntrospector,
                                              LoadBalancedRetryFactory loadBalancedRetryPolicyFactory) {
        super(delegate, config, serverIntrospector);
        this.loadBalancedRetryFactory = loadBalancedRetryPolicyFactory;
    }

    @Override
    public boolean isClientRetryable(ContextAwareRequest request) {
        return request!= null && isRequestRetryable(request);
    }

    private boolean isRequestRetryable(ContextAwareRequest request) {
        if (request.getContext() == null || request.getContext().getRetryable() == null) {
            return true;
        }
        return request.getContext().getRetryable();
    }

    private OkHttpRibbonResponse executeWithRetry(OkHttpRibbonRequest request, LoadBalancedRetryPolicy retryPolicy,
                                                  RetryCallback<OkHttpRibbonResponse, Exception> callback,
                                                  RecoveryCallback<OkHttpRibbonResponse> recoveryCallback) throws Exception {
        RetryTemplate retryTemplate = new RetryTemplate();
        BackOffPolicy backOffPolicy = loadBalancedRetryFactory.createBackOffPolicy(this.getClientName());
        retryTemplate.setBackOffPolicy(backOffPolicy == null ? new NoBackOffPolicy() : backOffPolicy);
        RetryListener[] retryListeners = this.loadBalancedRetryFactory.createRetryListeners(this.getClientName());
        if (retryListeners != null && retryListeners.length != 0) {
            retryTemplate.setListeners(retryListeners);
        }
        boolean retryable = isRequestRetryable(request);
        retryTemplate.setRetryPolicy(retryPolicy == null || !retryable ? new NeverRetryPolicy()
                : new RetryPolicy(request, retryPolicy, this, this.getClientName()));
        return retryTemplate.execute(callback, recoveryCallback);
    }

    @Override
    public OkHttpRibbonResponse execute(final OkHttpRibbonRequest ribbonRequest,
                                        final IClientConfig configOverride) throws Exception {
        final LoadBalancedRetryPolicy retryPolicy = loadBalancedRetryFactory.createRetryPolicy(this.getClientName(), this);
        RetryCallback<OkHttpRibbonResponse, Exception> retryCallback  = new RetryCallback<OkHttpRibbonResponse, Exception>() {
            @Override
            public OkHttpRibbonResponse doWithRetry(RetryContext context) throws Exception {
                //on retries the policy will choose the server and set it in the context
                //extract the server and update the request being made
                OkHttpRibbonRequest newRequest = ribbonRequest;
                if(context instanceof LoadBalancedRetryContext) {
                    ServiceInstance service = ((LoadBalancedRetryContext)context).getServiceInstance();
                    validateServiceInstance(service);
                    //Reconstruct the request URI using the host and port set in the retry context
                    newRequest = newRequest.withNewUri(new URI(service.getUri().getScheme(),
                            newRequest.getURI().getUserInfo(), service.getHost(), service.getPort(),
                            newRequest.getURI().getPath(), newRequest.getURI().getQuery(),
                            newRequest.getURI().getFragment()));
                }
                if (isSecure(configOverride)) {
                    final URI secureUri = UriComponentsBuilder.fromUri(newRequest.getUri())
                            .scheme("https").build().toUri();
                    newRequest = newRequest.withNewUri(secureUri);
                }
                OkHttpClient httpClient = getOkHttpClient(configOverride, secure);

                final Request request = newRequest.toRequest();
                Response response = httpClient.newCall(request).execute();
                if(retryPolicy.retryableStatusCode(response.code())) {
                    ResponseBody responseBody = response.peekBody(Integer.MAX_VALUE);
                    response.close();
                    throw new OkHttpStatusCodeException(RetryableOkHttpLoadBalancingClient.this.clientName,
                            response, responseBody, newRequest.getURI());
                }
                return new OkHttpRibbonResponse(response, newRequest.getUri());
            }
        };
        return this.executeWithRetry(ribbonRequest, retryPolicy, retryCallback, new LoadBalancedRecoveryCallback<OkHttpRibbonResponse, Response>(){

            @Override
            protected OkHttpRibbonResponse createResponse(Response response, URI uri) {
                return new OkHttpRibbonResponse(response, uri);
            }
        });
    }



    @Override
    public RequestSpecificRetryHandler getRequestSpecificRetryHandler(OkHttpRibbonRequest request, IClientConfig requestConfig) {
        return new RequestSpecificRetryHandler(false, false, RetryHandler.DEFAULT, null);
    }

    static class RetryPolicy extends InterceptorRetryPolicy {
        public RetryPolicy(HttpRequest request, LoadBalancedRetryPolicy policy, ServiceInstanceChooser serviceInstanceChooser, String serviceName) {
            super(request, policy, serviceInstanceChooser, serviceName);
        }
    }
}
  • 这里的executeWithRetry,也是通过设置RetryTemplate的RetryPolicy及BackOffPolicy,来执行重试策略

RetryTemplate

spring-retry-1.2.2.RELEASE-sources.jar!/org/springframework/retry/support/RetryTemplate.java

public class RetryTemplate implements RetryOperations {

    /**
     * Retry context attribute name that indicates the context should be considered global
     * state (never closed). TODO: convert this to a flag in the RetryState.
     */
    private static final String GLOBAL_STATE = "state.global";

    protected final Log logger = LogFactory.getLog(getClass());

    private volatile BackOffPolicy backOffPolicy = new NoBackOffPolicy();

    private volatile RetryPolicy retryPolicy = new SimpleRetryPolicy(3);

    private volatile RetryListener[] listeners = new RetryListener[0];

    private RetryContextCache retryContextCache = new MapRetryContextCache();

    private boolean throwLastExceptionOnExhausted;

    //......

    /**
     * Execute the callback once if the policy dictates that we can, otherwise execute the
     * recovery callback.
     * @param recoveryCallback the {@link RecoveryCallback}
     * @param retryCallback the {@link RetryCallback}
     * @param state the {@link RetryState}
     * @param <T> the type of the return value
     * @param <E> the exception type to throw
     * @see RetryOperations#execute(RetryCallback, RecoveryCallback, RetryState)
     * @throws ExhaustedRetryException if the retry has been exhausted.
     * @throws E an exception if the retry operation fails
     * @return T the retried value
     */
    protected <T, E extends Throwable> T doExecute(RetryCallback<T, E> retryCallback,
            RecoveryCallback<T> recoveryCallback, RetryState state)
            throws E, ExhaustedRetryException {

        RetryPolicy retryPolicy = this.retryPolicy;
        BackOffPolicy backOffPolicy = this.backOffPolicy;

        // Allow the retry policy to initialise itself...
        RetryContext context = open(retryPolicy, state);
        if (this.logger.isTraceEnabled()) {
            this.logger.trace("RetryContext retrieved: " + context);
        }

        // Make sure the context is available globally for clients who need
        // it...
        RetrySynchronizationManager.register(context);

        Throwable lastException = null;

        boolean exhausted = false;
        try {

            // Give clients a chance to enhance the context...
            boolean running = doOpenInterceptors(retryCallback, context);

            if (!running) {
                throw new TerminatedRetryException(
                        "Retry terminated abnormally by interceptor before first attempt");
            }

            // Get or Start the backoff context...
            BackOffContext backOffContext = null;
            Object resource = context.getAttribute("backOffContext");

            if (resource instanceof BackOffContext) {
                backOffContext = (BackOffContext) resource;
            }

            if (backOffContext == null) {
                backOffContext = backOffPolicy.start(context);
                if (backOffContext != null) {
                    context.setAttribute("backOffContext", backOffContext);
                }
            }

            /*
             * We allow the whole loop to be skipped if the policy or context already
             * forbid the first try. This is used in the case of external retry to allow a
             * recovery in handleRetryExhausted without the callback processing (which
             * would throw an exception).
             */
            while (canRetry(retryPolicy, context) && !context.isExhaustedOnly()) {

                try {
                    if (this.logger.isDebugEnabled()) {
                        this.logger.debug("Retry: count=" + context.getRetryCount());
                    }
                    // Reset the last exception, so if we are successful
                    // the close interceptors will not think we failed...
                    lastException = null;
                    return retryCallback.doWithRetry(context);
                }
                catch (Throwable e) {

                    lastException = e;

                    try {
                        registerThrowable(retryPolicy, state, context, e);
                    }
                    catch (Exception ex) {
                        throw new TerminatedRetryException("Could not register throwable",
                                ex);
                    }
                    finally {
                        doOnErrorInterceptors(retryCallback, context, e);
                    }

                    if (canRetry(retryPolicy, context) && !context.isExhaustedOnly()) {
                        try {
                            backOffPolicy.backOff(backOffContext);
                        }
                        catch (BackOffInterruptedException ex) {
                            lastException = e;
                            // back off was prevented by another thread - fail the retry
                            if (this.logger.isDebugEnabled()) {
                                this.logger
                                        .debug("Abort retry because interrupted: count="
                                                + context.getRetryCount());
                            }
                            throw ex;
                        }
                    }

                    if (this.logger.isDebugEnabled()) {
                        this.logger.debug(
                                "Checking for rethrow: count=" + context.getRetryCount());
                    }

                    if (shouldRethrow(retryPolicy, context, state)) {
                        if (this.logger.isDebugEnabled()) {
                            this.logger.debug("Rethrow in retry for policy: count="
                                    + context.getRetryCount());
                        }
                        throw RetryTemplate.<E>wrapIfNecessary(e);
                    }

                }

                /*
                 * A stateful attempt that can retry may rethrow the exception before now,
                 * but if we get this far in a stateful retry there's a reason for it,
                 * like a circuit breaker or a rollback classifier.
                 */
                if (state != null && context.hasAttribute(GLOBAL_STATE)) {
                    break;
                }
            }

            if (state == null && this.logger.isDebugEnabled()) {
                this.logger.debug(
                        "Retry failed last attempt: count=" + context.getRetryCount());
            }

            exhausted = true;
            return handleRetryExhausted(recoveryCallback, context, state);

        }
        catch (Throwable e) {
            throw RetryTemplate.<E>wrapIfNecessary(e);
        }
        finally {
            close(retryPolicy, context, state, lastException == null || exhausted);
            doCloseInterceptors(retryCallback, context, lastException);
            RetrySynchronizationManager.clear();
        }

    }

    //......
}
  • 整个retry的逻辑就是通过while控制重试次数,循环内没有返回,到循环外会抛出ExhaustedRetryException
  • BackOffPolicy的backOff方法通过sleeper.sleep执行重试间隔
  • BackOffPolicy有一个子接口SleepingBackOffPolicy,还有一个实现该接口的抽象类StatelessBackOffPolicy
  • SleepingBackOffPolicy的实现类有FixedBackOffPolicy、UniformRandomBackOffPolicy、ExponentialBackOffPolicy、ExponentialRandomBackOffPolicy这三个
  • StatelessBackOffPolicy有FixedBackOffPolicy、NoBackOffPolicy、UniformRandomBackOffPolicy三个子类

小结

ribbon的retry,有两个实现,一个是RetryableRibbonLoadBalancingHttpClient,另外一个是RetryableOkHttpLoadBalancingClient,他们使用的是spring-retry组件的RetryTemplate来实现,该template主要由RetryPolicy以及BackOffPolicy来控制重试。重试大的逻辑就是用while来控制重试次数,然后通过BackOffPolicy来控制重试时间间隔。

doc

  • 9. Retry

本文分享自微信公众号 - 码匠的流水账(geek_luandun)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2018-07-19

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 高性能网络通信框架Netty-Java NIO基础

    本节我们使用JDK中原生 NIO API来创建一个简单的TCP客户端与服务器交互的网络程序。

    加多
  • springboot整合mybatis(xml+注解)

    ​ 刚毕业的第一份工作是java开发,项目中需要用到mybatis,特此记录学习过程,这只是一个简单demo,mybatis用法很多不可能全部写出来,有更复杂的...

    用户2038589
  • CentOS安装配置

    用户3004328
  • Require.js

    通过require.js 可以对javascript文件进行别样引用 Require.js的使用: 第一步: 只需在html页面中引用require.js ...

    河湾欢儿
  • APP概括

    移动端:运行在移动设备上的产品 移动设备:手机 平板 我们做的页面只适用于安卓系统和ios系统,诺基亚是Windows Phone系统不用管

    河湾欢儿
  • spring基础(1:基本概念)

    本系列笔记来自对《Spring实战》第三版的整理,Spring版本为3.0   spring是为了解决企业级应用开发的复杂性而创建的,spring最根本的...

    用户2038589
  • 如何更改github工程的语言属性

    当创建github项目的时候,github本身会根据提交文件的数量来自动推断工程的开发语言,有时这种推断结果会与实际情况不太相符。

    Python疯子
  • JavaScript简单介绍

    基于事件和对象驱动,并具有安全性的语言。 该语言运行在浏览器端,也有在服务器端的名称为Node.js

    河湾欢儿
  • 高性能网络通信框架Netty-Netty客户端底层与Java NIO对应关系

    在讲解Netty客户端程序时候我们提到指定NioSocketChannel用于创建客户端NIO套接字通道的实例,下面我们来看NioSocketChannel是如...

    加多
  • 通过163smtp服务器向各大邮箱发送邮件(SOCKET编程)

    注意如果是163向163发送邮件,邮件的发送者(mail.sender,不是真实发送者这个必须和认证的用户名相同)可以设置为任意的,这样接收者收到邮件时会看到该...

    用户2038589

扫码关注云+社区

领取腾讯云代金券