Spring-Retry重试实现原理 | Alben's home

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      
13      
14      
15      
16      
17      
18      
19      
20      
21      
22      
23      
24      
25      
26      
27      
28      
29      
30      
31      
32      

@Configuration      
@EnableRetry      
public class Application {      
      
    @Bean      
    public RetryService retryService(){      
        return new RetryService();      
    }      
      
    public static void main(String[] args) throws Exception{      
        ApplicationContext applicationContext = new AnnotationConfigApplicationContext("springretry");      
        RetryService service1 = applicationContext.getBean("service", RetryService.class);      
        service1.service();      
    }      
}      
      
@Service("service")      
public class RetryService {      
      
    @Retryable(value = IllegalAccessException.class, maxAttempts = 5,      
            backoff= @Backoff(value = 1500, maxDelay = 100000, multiplier = 1.2))      
    public void service() throws IllegalAccessException {      
        System.out.println("service method...");      
        throw new IllegalAccessException("manual exception");      
    }      
      
    @Recover      
    public void recover(IllegalAccessException e){      
        System.out.println("service retry after Recover => " + e.getMessage());      
    }      
      
}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      
13      
14      
15      
16      

@Target(ElementType.TYPE)      
@Retention(RetentionPolicy.RUNTIME)      
@EnableAspectJAutoProxy(proxyTargetClass = false)      
@Import(RetryConfiguration.class)      
@Documented      
public @interface EnableRetry {      
      
	/**      
	 * Indicate whether subclass-based (CGLIB) proxies are to be created as opposed      
	 * to standard Java interface-based proxies. The default is {@code false}.      
	 *      
	 * @return whether to proxy or not to proxy the class      
	 */      
	boolean proxyTargetClass() default false;      
      
}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      

@PostConstruct      
	public void init() {      
		Set<Class<? extends Annotation>> retryableAnnotationTypes = new LinkedHashSet<Class<? extends Annotation>>(1);      
		retryableAnnotationTypes.add(Retryable.class);      
        //创建pointcut      
		this.pointcut = buildPointcut(retryableAnnotationTypes);      
        //创建advice      
		this.advice = buildAdvice();      
		if (this.advice instanceof BeanFactoryAware) {      
			((BeanFactoryAware) this.advice).setBeanFactory(beanFactory);      
		}      
	}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      
13      

protected Pointcut buildPointcut(Set<Class<? extends Annotation>> retryAnnotationTypes) {      
		ComposablePointcut result = null;      
		for (Class<? extends Annotation> retryAnnotationType : retryAnnotationTypes) {      
			Pointcut filter = new AnnotationClassOrMethodPointcut(retryAnnotationType);      
			if (result == null) {      
				result = new ComposablePointcut(filter);      
			}      
			else {      
				result.union(filter);      
			}      
		}      
		return result;      
	}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      
13      
14      
15      
16      
17      
18      
19      
20      
21      

//创建advice对象，即拦截器      
   protected Advice buildAdvice() {      
   	//下面关注这个对象      
	AnnotationAwareRetryOperationsInterceptor interceptor = new AnnotationAwareRetryOperationsInterceptor();      
	if (retryContextCache != null) {      
		interceptor.setRetryContextCache(retryContextCache);      
	}      
	if (retryListeners != null) {      
		interceptor.setListeners(retryListeners);      
	}      
	if (methodArgumentsKeyGenerator != null) {      
		interceptor.setKeyGenerator(methodArgumentsKeyGenerator);      
	}      
	if (newMethodArgumentsIdentifier != null) {      
		interceptor.setNewItemIdentifier(newMethodArgumentsIdentifier);      
	}      
	if (sleeper != null) {      
		interceptor.setSleeper(sleeper);      
	}      
	return interceptor;      
}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      

@Override      
	public Object invoke(MethodInvocation invocation) throws Throwable {      
		MethodInterceptor delegate = getDelegate(invocation.getThis(), invocation.getMethod());      
		if (delegate != null) {      
			return delegate.invoke(invocation);      
		}      
		else {      
			return invocation.proceed();      
		}      
	}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      
13      
14      
15      
16      
17      
18      
19      
20      
21      
22      
23      
24      
25      
26      
27      
28      
29      
30      
31      
32      
33      
34      
35      
36      
37      

private MethodInterceptor getDelegate(Object target, Method method) {      
		if (!this.delegates.containsKey(target) || !this.delegates.get(target).containsKey(method)) {      
			synchronized (this.delegates) {      
				if (!this.delegates.containsKey(target)) {      
					this.delegates.put(target, new HashMap<Method, MethodInterceptor>());      
				}      
				Map<Method, MethodInterceptor> delegatesForTarget = this.delegates.get(target);      
				if (!delegatesForTarget.containsKey(method)) {      
					Retryable retryable = AnnotationUtils.findAnnotation(method, Retryable.class);      
					if (retryable == null) {      
						retryable = AnnotationUtils.findAnnotation(method.getDeclaringClass(), Retryable.class);      
					}      
					if (retryable == null) {      
						retryable = findAnnotationOnTarget(target, method);      
					}      
					if (retryable == null) {      
						return delegatesForTarget.put(method, null);      
					}      
					MethodInterceptor delegate;      
					//支持自定义MethodInterceptor，而且优先级最高      
					if (StringUtils.hasText(retryable.interceptor())) {      
						delegate = this.beanFactory.getBean(retryable.interceptor(), MethodInterceptor.class);      
					}      
					else if (retryable.stateful()) {      
                    	//得到“有状态”的interceptor      
						delegate = getStatefulInterceptor(target, method, retryable);      
					}      
					else {      
                    	//得到“无状态”的interceptor      
						delegate = getStatelessInterceptor(target, method, retryable);      
					}      
					delegatesForTarget.put(method, delegate);      
				}      
			}      
		}      
		return this.delegates.get(target).get(method);      
	}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      
13      

private MethodInterceptor getStatelessInterceptor(Object target, Method method, Retryable retryable) {      
		//生成一个RetryTemplate      
		RetryTemplate template = createTemplate(retryable.listeners());      
		//生成retryPolicy      
		template.setRetryPolicy(getRetryPolicy(retryable));      
		//生成backoffPolicy      
		template.setBackOffPolicy(getBackoffPolicy(retryable.backoff()));      
		return RetryInterceptorBuilder.stateless()      
				.retryOperations(template)      
				.label(retryable.label())      
				.recoverer(getRecoverer(target, method))      
				.build();      
	}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      
13      
14      
15      
16      
17      
18      
19      
20      
21      
22      
23      
24      
25      
26      
27      
28      
29      
30      
31      
32      
33      
34      
35      
36      
37      
38      
39      
40      
41      
42      
43      
44      
45      
46      
47      
48      
49      
50      
51      
52      
53      
54      
55      

public Object invoke(final MethodInvocation invocation) throws Throwable {      
      
		String name;      
		if (StringUtils.hasText(label)) {      
			name = label;      
		} else {      
			name = invocation.getMethod().toGenericString();      
		}      
		final String label = name;      
      
		//定义了一个RetryCallback，其实看它的doWithRetry方法，调用了invocation的proceed()方法，是不是有点眼熟，这就是AOP的拦截链调用，如果没有拦截链，那就是对原来方法的调用。      
		RetryCallback<Object, Throwable> retryCallback = new RetryCallback<Object, Throwable>() {      
      
			public Object doWithRetry(RetryContext context) throws Exception {      
				      
				context.setAttribute(RetryContext.NAME, label);      
      
				/*      
				 * If we don't copy the invocation carefully it won't keep a reference to      
				 * the other interceptors in the chain. We don't have a choice here but to      
				 * specialise to ReflectiveMethodInvocation (but how often would another      
				 * implementation come along?).      
				 */      
				if (invocation instanceof ProxyMethodInvocation) {      
					try {      
						return ((ProxyMethodInvocation) invocation).invocableClone().proceed();      
					}      
					catch (Exception e) {      
						throw e;      
					}      
					catch (Error e) {      
						throw e;      
					}      
					catch (Throwable e) {      
						throw new IllegalStateException(e);      
					}      
				}      
				else {      
					throw new IllegalStateException(      
							"MethodInvocation of the wrong type detected - this should not happen with Spring AOP, " +      
									"so please raise an issue if you see this exception");      
				}      
			}      
      
		};      
      
		if (recoverer != null) {      
			ItemRecovererCallback recoveryCallback = new ItemRecovererCallback(      
					invocation.getArguments(), recoverer);      
			return this.retryOperations.execute(retryCallback, recoveryCallback);      
		}      
		//最终还是进入到retryOperations的execute方法，这个retryOperations就是在之前的builder set进来的RetryTemplate。      
		return this.retryOperations.execute(retryCallback);      
      
	}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      
13      
14      
15      
16      
17      
18      
19      
20      
21      
22      
23      
24      
25      
26      
27      
28      
29      
30      
31      
32      
33      
34      
35      
36      
37      
38      
39      
40      
41      
42      
43      
44      
45      
46      
47      
48      
49      
50      
51      
52      
53      
54      
55      
56      
57      
58      
59      
60      
61      
62      
63      
64      
65      
66      
67      
68      
69      
70      
71      
72      
73      
74      
75      
76      
77      
78      
79      
80      
81      
82      
83      
84      
85      
86      
87      
88      
89      
90      
91      
92      
93      
94      
95      
96      
97      
98      
99      
100      
101      
102      
103      
104      
105      
106      
107      
108      
109      
110      
111      
112      
113      
114      
115      
116      
117      
118      
119      
120      
121      
122      
123      
124      
125      
126      
127      
128      
129      
130      
131      
132      
133      
134      
135      
136      
137      
138      
139      

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;      
      
		//新建一个RetryContext来保存本轮重试的上下文      
		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 {      
      
			//如果有注册RetryListener，则会调用它的open方法，给调用者一个通知。      
			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);      
				}      
			}      
      
			//判断能否重试，就是调用RetryPolicy的canRetry方法来判断。      
			//这个循环会直到原方法不抛出异常，或不需要再重试      
			while (canRetry(retryPolicy, context) && !context.isExhaustedOnly()) {      
      
				try {      
					if (this.logger.isDebugEnabled()) {      
						this.logger.debug("Retry: count=" + context.getRetryCount());      
					}      
					//清除上次记录的异常      
					lastException = null;      
					//doWithRetry方法，一般来说就是原方法      
					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 {      
						//调用RetryListener的onError方法      
						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;      
			//重试结束后如果有兜底Recovery方法则执行，否则抛异常      
			return handleRetryExhausted(recoveryCallback, context, state);      
      
		}      
		catch (Throwable e) {      
			throw RetryTemplate.<E>wrapIfNecessary(e);      
		}      
		finally {      
			//处理一些关闭逻辑      
			close(retryPolicy, context, state, lastException == null || exhausted);      
			//调用RetryListener的close方法      
			doCloseInterceptors(retryCallback, context, lastException);      
			RetrySynchronizationManager.clear();      
		}      
      
	}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      
13      
14      
15      
16      
17      
18      
19      
20      
21      
22      
23      
24      
25      
26      
27      
28      
29      
30      
31      
32      
33      
34      
35      
36      
37      
38      
39      
40      
41      
42      

private RetryPolicy getRetryPolicy(Annotation retryable) {      
		Map<String, Object> attrs = AnnotationUtils.getAnnotationAttributes(retryable);      
		@SuppressWarnings("unchecked")      
		Class<? extends Throwable>[] includes = (Class<? extends Throwable>[]) attrs.get("value");      
		String exceptionExpression = (String) attrs.get("exceptionExpression");      
		boolean hasExpression = StringUtils.hasText(exceptionExpression);      
		if (includes.length == 0) {      
			@SuppressWarnings("unchecked")      
			Class<? extends Throwable>[] value = (Class<? extends Throwable>[]) attrs.get("include");      
			includes = value;      
		}      
		@SuppressWarnings("unchecked")      
		Class<? extends Throwable>[] excludes = (Class<? extends Throwable>[]) attrs.get("exclude");      
		Integer maxAttempts = (Integer) attrs.get("maxAttempts");      
		String maxAttemptsExpression = (String) attrs.get("maxAttemptsExpression");      
		if (StringUtils.hasText(maxAttemptsExpression)) {      
			maxAttempts = PARSER.parseExpression(resolve(maxAttemptsExpression), PARSER_CONTEXT)      
					.getValue(this.evaluationContext, Integer.class);      
		}      
		if (includes.length == 0 && excludes.length == 0) {      
			SimpleRetryPolicy simple = hasExpression ? new ExpressionRetryPolicy(resolve(exceptionExpression))      
															.withBeanFactory(this.beanFactory)      
													 : new SimpleRetryPolicy();      
			simple.setMaxAttempts(maxAttempts);      
			return simple;      
		}      
		Map<Class<? extends Throwable>, Boolean> policyMap = new HashMap<Class<? extends Throwable>, Boolean>();      
		for (Class<? extends Throwable> type : includes) {      
			policyMap.put(type, true);      
		}      
		for (Class<? extends Throwable> type : excludes) {      
			policyMap.put(type, false);      
		}      
		boolean retryNotExcluded = includes.length == 0;      
		if (hasExpression) {      
			return new ExpressionRetryPolicy(maxAttempts, policyMap, true, exceptionExpression, retryNotExcluded)      
					.withBeanFactory(this.beanFactory);      
		}      
		else {      
			return new SimpleRetryPolicy(maxAttempts, policyMap, true, retryNotExcluded);      
		}      
	}      

1      
2      
3      
4      
5      
6      
7      
8      
9      
10      
11      
12      
13      
14      
15      
16      
17      
18      
19      
20      
21      
22      
23      
24      
25      
26      
27      
28      
29      
30      
31      
32      
33      
34      
35      
36      
37      
38      
39      
40      
41      
42      
43      
44      
45      

private BackOffPolicy getBackoffPolicy(Backoff backoff) {      
		long min = backoff.delay() == 0 ? backoff.value() : backoff.delay();      
		if (StringUtils.hasText(backoff.delayExpression())) {      
			min = PARSER.parseExpression(resolve(backoff.delayExpression()), PARSER_CONTEXT)      
					.getValue(this.evaluationContext, Long.class);      
		}      
		long max = backoff.maxDelay();      
		if (StringUtils.hasText(backoff.maxDelayExpression())) {      
			max = PARSER.parseExpression(resolve(backoff.maxDelayExpression()), PARSER_CONTEXT)      
					.getValue(this.evaluationContext, Long.class);      
		}      
		double multiplier = backoff.multiplier();      
		if (StringUtils.hasText(backoff.multiplierExpression())) {      
			multiplier = PARSER.parseExpression(resolve(backoff.multiplierExpression()), PARSER_CONTEXT)      
					.getValue(this.evaluationContext, Double.class);      
		}      
		if (multiplier > 0) {      
			ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();      
			if (backoff.random()) {      
				policy = new ExponentialRandomBackOffPolicy();      
			}      
			policy.setInitialInterval(min);      
			policy.setMultiplier(multiplier);      
			policy.setMaxInterval(max > min ? max : ExponentialBackOffPolicy.DEFAULT_MAX_INTERVAL);      
			if (this.sleeper != null) {      
				policy.setSleeper(this.sleeper);      
			}      
			return policy;      
		}      
		if (max > min) {      
			UniformRandomBackOffPolicy policy = new UniformRandomBackOffPolicy();      
			policy.setMinBackOffPeriod(min);      
			policy.setMaxBackOffPeriod(max);      
			if (this.sleeper != null) {      
				policy.setSleeper(this.sleeper);      
			}      
			return policy;      
		}      
		FixedBackOffPolicy policy = new FixedBackOffPolicy();      
		policy.setBackOffPeriod(min);      
		if (this.sleeper != null) {      
			policy.setSleeper(this.sleeper);      
		}      
		return policy;      
	}      

1      
2      
3      
4      
5      
6      
7      

@Override      
	public boolean canRetry(RetryContext context) {      
		Throwable t = context.getLastThrowable();      
		//判断抛出的异常是否符合重试的异常      
		//还有，是否超过了重试的次数      
		return (t == null || retryForException(t)) && context.getRetryCount() < maxAttempts;      
	}      

1      
2      
3      
4      
5      
6      
7      
8      
9      

protected void doBackOff() throws BackOffInterruptedException {      
		try {      
			//就是sleep固定的时间      
			sleeper.sleep(backOffPeriod);      
		}      
		catch (InterruptedException e) {      
			throw new BackOffInterruptedException("Thread interrupted while sleeping", e);      
		}      
	}