聊聊feign的Retryer

序

本文主要研究一下 feign 的 Retryer

Retryer

feign-core-10.2.3-sources.jar!/feign/Retryer.java

public interface Retryer extends Cloneable {

  /**
   * if retry is permitted, return (possibly after sleeping). Otherwise propagate the exception.
   */
  void continueOrPropagate(RetryableException e);

  Retryer clone();

  class Default implements Retryer {

    private final int maxAttempts;
    private final long period;
    private final long maxPeriod;
    int attempt;
    long sleptForMillis;

    public Default() {this(100, SECONDS.toMillis(1), 5);
    }

    public Default(long period, long maxPeriod, int maxAttempts) {
      this.period = period;
      this.maxPeriod = maxPeriod;
      this.maxAttempts = maxAttempts;
      this.attempt = 1;
    }

    // visible for testing;
    protected long currentTimeMillis() {return System.currentTimeMillis();
    }

    public void continueOrPropagate(RetryableException e) {if (attempt++ >= maxAttempts) {throw e;}

      long interval;
      if (e.retryAfter() != null) {interval = e.retryAfter().getTime() - currentTimeMillis();
        if (interval > maxPeriod) {interval = maxPeriod;}
        if (interval < 0) {return;}
      } else {interval = nextMaxInterval();
      }
      try {Thread.sleep(interval);
      } catch (InterruptedException ignored) {Thread.currentThread().interrupt();
        throw e;
      }
      sleptForMillis += interval;
    }

    /**
     * Calculates the time interval to a retry attempt. <br>
     * The interval increases exponentially with each attempt, at a rate of nextInterval *= 1.5
     * (where 1.5 is the backoff factor), to the maximum interval.
     *
     * @return time in nanoseconds from now until the next attempt.
     */
    long nextMaxInterval() {long interval = (long) (period * Math.pow(1.5, attempt - 1));
      return interval > maxPeriod ? maxPeriod : interval;
    }

    @Override
    public Retryer clone() {return new Default(period, maxPeriod, maxAttempts);
    }
  }

  /**
   * Implementation that never retries request. It propagates the RetryableException.
   */
  Retryer NEVER_RETRY = new Retryer() {

    @Override
    public void continueOrPropagate(RetryableException e) {throw e;}

    @Override
    public Retryer clone() {return this;}
  };
}

• Retryer 继承了 Cloneable 接口，它定义了 continueOrPropagate、clone 方法；它内置了一个名为 Default 以及名为 NEVER_RETRY 的实现
• Default 有 period、maxPeriod、maxAttempts 参数可以设置，默认构造器使用的 period 为 100，maxPeriod 为 1000，maxAttempts 为 5；continueOrPropagate 方法首先判断 attempt 是否达到阈值，达到则抛出异常，否则进一步计算 interval，然后进行 sleep
• NEVER_RETRY 的 continueOrPropagate 直接抛出异常，而 clone 方法直接返回当前实例

SynchronousMethodHandler

feign-core-10.2.3-sources.jar!/feign/SynchronousMethodHandler.java

final class SynchronousMethodHandler implements MethodHandler {
    //......

  public Object invoke(Object[] argv) throws Throwable {RequestTemplate template = buildTemplateFromArgs.create(argv);
    Retryer retryer = this.retryer.clone();
    while (true) {
      try {return executeAndDecode(template);
      } catch (RetryableException e) {
        try {retryer.continueOrPropagate(e);
        } catch (RetryableException th) {Throwable cause = th.getCause();
          if (propagationPolicy == UNWRAP && cause != null) {throw cause;} else {throw th;}
        }
        if (logLevel != Logger.Level.NONE) {logger.logRetry(metadata.configKey(), logLevel);
        }
        continue;
      }
    }
  }


    //......
}

• SynchronousMethodHandler 的 invoke 的方法首先使用 retryer.clone() 创建一个 retryer，然后在捕获到 RetryableException 的时候，会执行 retryer.continueOrPropagate(e)

RetryableException

feign-core-10.2.3-sources.jar!/feign/RetryableException.java

public class RetryableException extends FeignException {

  private static final long serialVersionUID = 1L;

  private final Long retryAfter;
  private final HttpMethod httpMethod;

  /**
   * @param retryAfter usually corresponds to the {@link feign.Util#RETRY_AFTER} header.
   */
  public RetryableException(int status, String message, HttpMethod httpMethod, Throwable cause,
      Date retryAfter) {super(status, message, cause);
    this.httpMethod = httpMethod;
    this.retryAfter = retryAfter != null ? retryAfter.getTime() : null;}

  /**
   * @param retryAfter usually corresponds to the {@link feign.Util#RETRY_AFTER} header.
   */
  public RetryableException(int status, String message, HttpMethod httpMethod, Date retryAfter) {super(status, message);
    this.httpMethod = httpMethod;
    this.retryAfter = retryAfter != null ? retryAfter.getTime() : null;}

  /**
   * Sometimes corresponds to the {@link feign.Util#RETRY_AFTER} header present in {@code 503}
   * status. Other times parsed from an application-specific response. Null if unknown.
   */
  public Date retryAfter() {return retryAfter != null ? new Date(retryAfter) : null;
  }

  public HttpMethod method() {return this.httpMethod;}
}

• RetryableException 继承了 FeignException，它的构造器会接收 retryAfter，该参数可以为 null

FeignException

feign-core-10.2.3-sources.jar!/feign/FeignException.java

public class FeignException extends RuntimeException {
    //......

  static FeignException errorReading(Request request, Response response, IOException cause) {
    return new FeignException(response.status(),
        format("%s reading %s %s", cause.getMessage(), request.httpMethod(), request.url()),
        cause,
        request.requestBody().asBytes());
  }

    //......
}

• FeignException 定义了 errorReading 静态方法，它创建的是 FeignException

ErrorDecoder

feign-core-10.2.3-sources.jar!/feign/codec/ErrorDecoder.java

public interface ErrorDecoder {
    //......

  public static class Default implements ErrorDecoder {private final RetryAfterDecoder retryAfterDecoder = new RetryAfterDecoder();

    @Override
    public Exception decode(String methodKey, Response response) {FeignException exception = errorStatus(methodKey, response);
      Date retryAfter = retryAfterDecoder.apply(firstOrNull(response.headers(), RETRY_AFTER));
      if (retryAfter != null) {
        return new RetryableException(response.status(),
            exception.getMessage(),
            response.request().httpMethod(),
            exception,
            retryAfter);
      }
      return exception;
    }

    private <T> T firstOrNull(Map<String, Collection<T>> map, String key) {if (map.containsKey(key) && !map.get(key).isEmpty()) {return map.get(key).iterator().next();
      }
      return null;
    }
  }

  static class RetryAfterDecoder {

    static final DateFormat RFC822_FORMAT =
        new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss'GMT'", US);
    private final DateFormat rfc822Format;

    RetryAfterDecoder() {this(RFC822_FORMAT);
    }

    RetryAfterDecoder(DateFormat rfc822Format) {this.rfc822Format = checkNotNull(rfc822Format, "rfc822Format");
    }

    protected long currentTimeMillis() {return System.currentTimeMillis();
    }

    /**
     * returns a date that corresponds to the first time a request can be retried.
     *
     * @param retryAfter String in
     *        <a href="https://tools.ietf.org/html/rfc2616#section-14.37" >Retry-After format</a>
     */
    public Date apply(String retryAfter) {if (retryAfter == null) {return null;}
      if (retryAfter.matches("^[0-9]+$")) {long deltaMillis = SECONDS.toMillis(Long.parseLong(retryAfter));
        return new Date(currentTimeMillis() + deltaMillis);
      }
      synchronized (rfc822Format) {
        try {return rfc822Format.parse(retryAfter);
        } catch (ParseException ignored) {return null;}
      }
    }
  }

    //......
}

• ErrorDecoder 提供了 Default 的默认实现，其 decode 方法会使用 RetryAfterDecoder 来计算 retryAfter 值，在该值不为 null 时会返回 RetryableException；RetryAfterDecoder 的 apply 方法会根据 retryAfter 来计算 retryAfter 日期，该 retryAfter 参数是从 response 的名为 Retry-After 的 header 读取而来

小结

• Retryer 继承了 Cloneable 接口，它定义了 continueOrPropagate、clone 方法；它内置了一个名为 Default 以及名为 NEVER_RETRY 的实现
• Default 有 period、maxPeriod、maxAttempts 参数可以设置，默认构造器使用的 period 为 100，maxPeriod 为 1000，maxAttempts 为 5；continueOrPropagate 方法首先判断 attempt 是否达到阈值，达到则抛出异常，否则进一步计算 interval，然后进行 sleep
• NEVER_RETRY 的 continueOrPropagate 直接抛出异常，而 clone 方法直接返回当前实例

doc

• Retryer