警惕看不見的重試機制：為什么使用RPC必須考慮冪等性
JAVA前線
歡迎大家關(guān)注公眾號「JAVA前線」查看更多精彩分享，主要包括源碼分析、實際應(yīng)用、架構(gòu)思維、職場分享、產(chǎn)品思考等等，同時也非常歡迎大家加我微信「java_front」一起交流學(xué)習(xí)
0 文章概述
在RPC場景中因為重試或者沒有實現(xiàn)冪等性而導(dǎo)致的重復(fù)數(shù)據(jù)問題，必須引起大家重視，有可能會造成例如一次購買創(chuàng)建多筆訂單，一條通知信息被發(fā)送多次等問題，這是技術(shù)人員必須面對和解決的問題。
有人可能會說：當調(diào)用失敗時程序并沒有顯示重試，為什么還會產(chǎn)生重復(fù)數(shù)據(jù)問題呢？這是因為即使沒有顯示重試，RPC框架在集群容錯機制中自動進行了重試，這個問題必須引起關(guān)注。
本文我們以DUBBO框架為例分析為什么重試，怎么做重試，怎么做冪等三個問題。
1 為什么重試
如果簡單對一個RPC交互過程進行分類，可以分為三類：響應(yīng)成功、響應(yīng)失敗、沒有響應(yīng)。
對于響應(yīng)成功和響應(yīng)失敗兩種情況，消費者很好處理。因為響應(yīng)信息明確，所以只要根據(jù)響應(yīng)信息，繼續(xù)處理成功或者失敗邏輯即可。但是沒有響應(yīng)這種場景比較難處理，這是因為沒有響應(yīng)可能包含以下情況：
(1) 生產(chǎn)者根本沒有接收到請求
(2) 生產(chǎn)者接收到請求并且已處理成功，但是消費者沒有接收到響應(yīng)
(3) 生產(chǎn)者接收到請求并且已處理失敗，但是消費者沒有接收到響應(yīng)
假設(shè)你是一名RPC框架設(shè)計者，究竟是選擇重試還是放棄調(diào)用呢？其實最終如何選擇取決于業(yè)務(wù)特性，有的業(yè)務(wù)本身就具有冪等性，但是有的業(yè)務(wù)不能允許重試否則會造成重復(fù)數(shù)據(jù)。
那么誰對業(yè)務(wù)特性最熟悉呢？答案是消費者，因為消費者作為調(diào)用方肯定最熟悉自身業(yè)務(wù)，所以RPC框架只要提供一些策略供消費者選擇即可。
2 怎么做重試
2.1 集群容錯策略
DUBBO作為一款優(yōu)秀RPC框架，提供了如下集群容錯策略供消費者選擇：
Failover： 故障轉(zhuǎn)移
Failfast： 快速失敗
Failsafe： 安全失敗
Failback： 異步重試
Forking： 并行調(diào)用
Broadcast：廣播調(diào)用
(1) Failover
故障轉(zhuǎn)移策略。作為默認策略，當消費發(fā)生異常時，通過負載均衡策略再選擇一個生產(chǎn)者節(jié)點進行調(diào)用，直到達到重試次數(shù)
(2) Failfast
快速失敗策略。消費者只消費一次服務(wù)，當發(fā)生異常時則直接拋出
(3) Failsafe
安全失敗策略。消費者只消費一次服務(wù)，如果消費失敗則包裝一個空結(jié)果，不拋出異常
(4) Failback
異步重試策略。消費發(fā)生異常時返回一個空結(jié)果，失敗請求將會進行異步重試。如果重試超過最大重試次數(shù)還不成功，放棄重試并不拋出異常
(5) Forking
并行調(diào)用策略。消費者通過線程池并發(fā)調(diào)用多個生產(chǎn)者，只要有一個成功就算成功
(6) Broadcast
廣播調(diào)用策略。消費者遍歷調(diào)用所有生產(chǎn)者節(jié)點，任何一個出現(xiàn)異常則拋出異常
2.2 源碼分析
2.2.1 Failover
Failover故障轉(zhuǎn)移策略作為默認策略，當消費發(fā)生異常時，通過負載均衡策略再選擇一個生產(chǎn)者節(jié)點進行調(diào)用，直到達到重試次數(shù)。即使業(yè)務(wù)代碼沒有顯示重試，也有可能多次執(zhí)行消費邏輯從而造成重復(fù)數(shù)據(jù)：
public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
public FailoverClusterInvoker(Directory<T> directory) {
super(directory);
}
@Override
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
// 所有生產(chǎn)者Invokers
List<Invoker<T>> copyInvokers = invokers;
checkInvokers(copyInvokers, invocation);
String methodName = RpcUtils.getMethodName(invocation);
// 獲取重試次數(shù)
int len = getUrl().getMethodParameter(methodName, Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
if (len <= 0) {
len = 1;
}
RpcException le = null;
// 已經(jīng)調(diào)用過的生產(chǎn)者
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyInvokers.size());
Set<String> providers = new HashSet<String>(len);
// 重試直到達到最大次數(shù)
for (int i = 0; i < len; i++) {
if (i > 0) {
// 如果當前實例被銷毀則拋出異常
checkWhetherDestroyed();
// 根據(jù)路由策略選出可用生產(chǎn)者Invokers
copyInvokers = list(invocation);
// 重新檢查
checkInvokers(copyInvokers, invocation);
}
// 負載均衡選擇一個生產(chǎn)者Invoker
Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
invoked.add(invoker);
RpcContext.getContext().setInvokers((List) invoked);
try {
// 服務(wù)消費發(fā)起遠程調(diào)用
Result result = invoker.invoke(invocation);
if (le != null && logger.isWarnEnabled()) {
logger.warn("Although retry the method " + methodName + " in the service " + getInterface().getName() + " was successful by the provider " + invoker.getUrl().getAddress() + ", but there have been failed providers " + providers + " (" + providers.size() + "/" + copyInvokers.size() + ") from the registry " + directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version " + Version.getVersion() + ". Last error is: " + le.getMessage(), le);
}
// 有結(jié)果則返回
return result;
} catch (RpcException e) {
// 業(yè)務(wù)異常直接拋出
if (e.isBiz()) {
throw e;
}
le = e;
} catch (Throwable e) {
// RpcException不拋出繼續(xù)重試
le = new RpcException(e.getMessage(), e);
} finally {
// 保存已經(jīng)訪問過的生產(chǎn)者
providers.add(invoker.getUrl().getAddress());
}
}
throw new RpcException(le.getCode(), "Failed to invoke the method " + methodName + " in the service " + getInterface().getName() + ". Tried " + len + " times of the providers " + providers + " (" + providers.size() + "/" + copyInvokers.size() + ") from the registry " + directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version " + Version.getVersion() + ". Last error is: " + le.getMessage(), le.getCause() != null ? le.getCause() : le);
}
}
消費者調(diào)用生產(chǎn)者節(jié)點A發(fā)生RpcException異常時（例如超時異常），在未達到最大重試次數(shù)之前，消費者會通過負載均衡策略再次選擇其它生產(chǎn)者節(jié)點消費。試想如果生產(chǎn)者節(jié)點A其實已經(jīng)處理成功了，但是沒有及時將成功結(jié)果返回給消費者，那么再次重試就可能造成重復(fù)數(shù)據(jù)問題。
2.2.2 Failfast
快速失敗策略。消費者只消費一次服務(wù)，當發(fā)生異常時則直接拋出，不會進行重試：
public class FailfastClusterInvoker<T> extends AbstractClusterInvoker<T> {
public FailfastClusterInvoker(Directory<T> directory) {
super(directory);
}
@Override
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
// 檢查生產(chǎn)者Invokers是否合法
checkInvokers(invokers, invocation);
// 負載均衡選擇一個生產(chǎn)者Invoker
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
try {
// 服務(wù)消費發(fā)起遠程調(diào)用
return invoker.invoke(invocation);
} catch (Throwable e) {
// 服務(wù)消費失敗不重試直接拋出異常
if (e instanceof RpcException && ((RpcException) e).isBiz()) {
throw (RpcException) e;
}
throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0,
"Failfast invoke providers " + invoker.getUrl() + " " + loadbalance.getClass().getSimpleName()
+ " select from all providers " + invokers + " for service " + getInterface().getName()
+ " method " + invocation.getMethodName() + " on consumer " + NetUtils.getLocalHost()
+ " use dubbo version " + Version.getVersion()
+ ", but no luck to perform the invocation. Last error is: " + e.getMessage(),
e.getCause() != null ? e.getCause() : e);
}
}
}
2.2.3 Failsafe
安全失敗策略。消費者只消費一次服務(wù)，如果消費失敗則包裝一個空結(jié)果，不拋出異常，不會進行重試：
public class FailsafeClusterInvoker<T> extends AbstractClusterInvoker<T> {
private static final Logger logger = LoggerFactory.getLogger(FailsafeClusterInvoker.class);
public FailsafeClusterInvoker(Directory<T> directory) {
super(directory);
}
@Override
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
// 檢查生產(chǎn)者Invokers是否合法
checkInvokers(invokers, invocation);
// 負載均衡選擇一個生產(chǎn)者Invoker
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
// 服務(wù)消費發(fā)起遠程調(diào)用
return invoker.invoke(invocation);
} catch (Throwable e) {
// 消費失敗包裝為一個空結(jié)果對象
logger.error("Failsafe ignore exception: " + e.getMessage(), e);
return new RpcResult();
}
}
}
2.2.4 Failback
異步重試策略。消費發(fā)生異常時返回一個空結(jié)果，失敗請求會進行異步重試。如果重試超過最大重試次數(shù)還不成功，放棄重試并不拋出異常：
public class FailbackClusterInvoker<T> extends AbstractClusterInvoker<T> {
private static final Logger logger = LoggerFactory.getLogger(FailbackClusterInvoker.class);
private static final long RETRY_FAILED_PERIOD = 5;
private final int retries;
private final int failbackTasks;
private volatile Timer failTimer;
public FailbackClusterInvoker(Directory<T> directory) {
super(directory);
int retriesConfig = getUrl().getParameter(Constants.RETRIES_KEY, Constants.DEFAULT_FAILBACK_TIMES);
if (retriesConfig <= 0) {
retriesConfig = Constants.DEFAULT_FAILBACK_TIMES;
}
int failbackTasksConfig = getUrl().getParameter(Constants.FAIL_BACK_TASKS_KEY, Constants.DEFAULT_FAILBACK_TASKS);
if (failbackTasksConfig <= 0) {
failbackTasksConfig = Constants.DEFAULT_FAILBACK_TASKS;
}
retries = retriesConfig;
failbackTasks = failbackTasksConfig;
}
private void addFailed(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, Invoker<T> lastInvoker) {
if (failTimer == null) {
synchronized (this) {
if (failTimer == null) {
// 創(chuàng)建定時器
failTimer = new HashedWheelTimer(new NamedThreadFactory("failback-cluster-timer", true), 1, TimeUnit.SECONDS, 32, failbackTasks);
}
}
}
// 構(gòu)造定時任務(wù)
RetryTimerTask retryTimerTask = new RetryTimerTask(loadbalance, invocation, invokers, lastInvoker, retries, RETRY_FAILED_PERIOD);
try {
// 定時任務(wù)放入定時器等待執(zhí)行
failTimer.newTimeout(retryTimerTask, RETRY_FAILED_PERIOD, TimeUnit.SECONDS);
} catch (Throwable e) {
logger.error("Failback background works error,invocation->" + invocation + ", exception: " + e.getMessage());
}
}
@Override
protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
Invoker<T> invoker = null;
try {
// 檢查生產(chǎn)者Invokers是否合法
checkInvokers(invokers, invocation);
// 負責均衡選擇一個生產(chǎn)者Invoker
invoker = select(loadbalance, invocation, invokers, null);
// 消費服務(wù)發(fā)起遠程調(diào)用
return invoker.invoke(invocation);
} catch (Throwable e) {
logger.error("Failback to invoke method " + invocation.getMethodName() + ", wait for retry in background. Ignored exception: " + e.getMessage() + ", ", e);
// 如果服務(wù)消費失敗則記錄失敗請求
addFailed(loadbalance, invocation, invokers, invoker);
// 返回空結(jié)果
return new RpcResult();
}
}
@Override
public void destroy() {
super.destroy();
if (failTimer != null) {
failTimer.stop();
}
}
/**
* RetryTimerTask
*/
private class RetryTimerTask implements TimerTask {
private final Invocation invocation;
private final LoadBalance loadbalance;
private final List<Invoker<T>> invokers;
private final int retries;
private final long tick;
private Invoker<T> lastInvoker;
private int retryTimes = 0;
RetryTimerTask(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, Invoker<T> lastInvoker, int retries, long tick) {
this.loadbalance = loadbalance;
this.invocation = invocation;
this.invokers = invokers;
this.retries = retries;
this.tick = tick;
this.lastInvoker = lastInvoker;
}
@Override
public void run(Timeout timeout) {
try {
// 負載均衡選擇一個生產(chǎn)者Invoker
Invoker<T> retryInvoker = select(loadbalance, invocation, invokers, Collections.singletonList(lastInvoker));
lastInvoker = retryInvoker;
// 服務(wù)消費發(fā)起遠程調(diào)用
retryInvoker.invoke(invocation);
} catch (Throwable e) {
logger.error("Failed retry to invoke method " + invocation.getMethodName() + ", waiting again.", e);
// 超出最大重試次數(shù)記錄日志不拋出異常
if ((++retryTimes) >= retries) {
logger.error("Failed retry times exceed threshold (" + retries + "), We have to abandon, invocation->" + invocation);
} else {
// 未超出最大重試次數(shù)重新放入定時器
rePut(timeout);
}
}
}
private void rePut(Timeout timeout) {
if (timeout == null) {
return;
}
Timer timer = timeout.timer();
if (timer.isStop() || timeout.isCancelled()) {
return;
}
timer.newTimeout(timeout.task(), tick, TimeUnit.SECONDS);
}
}
}
2.2.5 Forking
并行調(diào)用策略。消費者通過線程池并發(fā)調(diào)用多個生產(chǎn)者，只要有一個成功就算成功：
public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> {
private final ExecutorService executor = Executors.newCachedThreadPool(new NamedInternalThreadFactory("forking-cluster-timer", true));
public ForkingClusterInvoker(Directory<T> directory) {
super(directory);
}
@Override
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
final List<Invoker<T>> selected;
// 獲取配置參數(shù)
final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
// 獲取并行執(zhí)行的Invoker列表
if (forks <= 0 || forks >= invokers.size()) {
selected = invokers;
} else {
selected = new ArrayList<>();
for (int i = 0; i < forks; i++) {
// 選擇生產(chǎn)者
Invoker<T> invoker = select(loadbalance, invocation, invokers, selected);
// 防止重復(fù)增加Invoker
if (!selected.contains(invoker)) {
selected.add(invoker);
}
}
}
RpcContext.getContext().setInvokers((List) selected);
final AtomicInteger count = new AtomicInteger();
final BlockingQueue<Object> ref = new LinkedBlockingQueue<>();
for (final Invoker<T> invoker : selected) {
// 在線程池中并發(fā)執(zhí)行
executor.execute(new Runnable() {
@Override
public void run() {
try {
// 執(zhí)行消費邏輯
Result result = invoker.invoke(invocation);
// 存儲消費結(jié)果
ref.offer(result);
} catch (Throwable e) {
// 如果異常次數(shù)大于等于forks參數(shù)值說明全部調(diào)用失敗，則把異常放入隊列
int value = count.incrementAndGet();
if (value >= selected.size()) {
ref.offer(e);
}
}
}
});
}
try {
// 從隊列獲取結(jié)果
Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
// 如果異常類型表示全部調(diào)用失敗則拋出異常
if (ret instanceof Throwable) {
Throwable e = (Throwable) ret;
throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0, "Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e);
}
return (Result) ret;
} catch (InterruptedException e) {
throw new RpcException("Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e);
}
} finally {
RpcContext.getContext().clearAttachments();
}
}
}
2.2.6 Broadcast
廣播調(diào)用策略。消費者遍歷調(diào)用所有生產(chǎn)者節(jié)點，任何一個出現(xiàn)異常則拋出異常：
public class BroadcastClusterInvoker<T> extends AbstractClusterInvoker<T> {
private static final Logger logger = LoggerFactory.getLogger(BroadcastClusterInvoker.class);
public BroadcastClusterInvoker(Directory<T> directory) {
super(directory);
}
@Override
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
RpcContext.getContext().setInvokers((List) invokers);
RpcException exception = null;
Result result = null;
// 遍歷調(diào)用所有生產(chǎn)者節(jié)點
for (Invoker<T> invoker : invokers) {
try {
// 執(zhí)行消費邏輯
result = invoker.invoke(invocation);
} catch (RpcException e) {
exception = e;
logger.warn(e.getMessage(), e);
} catch (Throwable e) {
exception = new RpcException(e.getMessage(), e);
logger.warn(e.getMessage(), e);
}
}
// 任何一個出現(xiàn)異常則拋出異常
if (exception != null) {
throw exception;
}
return result;
}
}
3 怎么做冪等
經(jīng)過上述分析我們知道，RPC框架自帶的重試機制可能會造成數(shù)據(jù)重復(fù)問題，那么在使用中必須考慮冪等性。冪等性是指一次操作與多次操作產(chǎn)生結(jié)果相同，并不會因為多次操作而產(chǎn)生不一致性。常見冪等方案有取消重試、冪等表、數(shù)據(jù)庫鎖、狀態(tài)機這些方案。
3.1 取消重試
取消重試有兩種方法，第一是設(shè)置重試次數(shù)為零，第二是選擇不重試的集群容錯策略。
<!-- 設(shè)置重試次數(shù)為零 -->
<dubbo:reference id="helloService" interface="com.java.front.dubbo.demo.provider.HelloService" retries="0" />
<!-- 選擇集群容錯方案 -->
<dubbo:reference id="helloService" interface="com.java.front.dubbo.demo.provider.HelloService" cluster="failfast" />
3.2 冪等表
假設(shè)用戶支付成功后，支付系統(tǒng)將支付成功消息，發(fā)送至消息隊列。物流系統(tǒng)訂閱到這個消息，準備為這筆訂單創(chuàng)建物流單。
但是消息隊列可能會重復(fù)推送，物流系統(tǒng)有可能接收到多次這條消息。我們希望達到效果：無論接收到多少條重復(fù)消息，只能創(chuàng)建一筆物流單。
解決方案是冪等表方案。新建一張冪等表，該表就是用來做冪等，無其它業(yè)務(wù)意義，有一個字段名為key建有唯一索引，這個字段是冪等標準。
物流系統(tǒng)訂閱到消息后，首先嘗試插入冪等表，訂單編號作為key字段。如果成功則繼續(xù)創(chuàng)建物流單，如果訂單編號已經(jīng)存在則違反唯一性原則，無法插入成功，說明已經(jīng)進行過業(yè)務(wù)處理，丟棄消息。
這張表數(shù)據(jù)量會比較大，我們可以通過定時任務(wù)對數(shù)據(jù)進行歸檔，例如只保留7天數(shù)據(jù)，其它數(shù)據(jù)存入歸檔表。
還有一種廣義冪等表就是我們可以用Redis替代數(shù)據(jù)庫，在創(chuàng)建物流單之前，我們可以檢查Redis是否存在該訂單編號數(shù)據(jù)，同時可以為這類數(shù)據(jù)設(shè)置7天過期時間。
3.3 狀態(tài)機
物流單創(chuàng)建成功后會發(fā)送消息，訂單系統(tǒng)訂閱到消息后更新狀態(tài)為完成，假設(shè)變更是將訂單狀態(tài)0更新至狀態(tài)1。訂單系統(tǒng)也可能收到多條消息，可能在狀態(tài)已經(jīng)被更新至狀態(tài)1之后，依然收到物流單創(chuàng)建成功消息。
解決方案是狀態(tài)機方案。首先繪制狀態(tài)機圖，分析狀態(tài)流轉(zhuǎn)形態(tài)。例如經(jīng)過分析狀態(tài)1已經(jīng)是最終態(tài)，那么即使接收到物流單創(chuàng)建成功消息也不再處理，丟棄消息。
3.4 數(shù)據(jù)庫鎖
數(shù)據(jù)庫鎖又可以分為悲觀鎖和樂觀鎖兩種類型，悲觀鎖是在獲取數(shù)據(jù)時加鎖：
select * from table where col='xxx' for update
樂觀鎖是在更新時加鎖，第一步首先查出數(shù)據(jù)，數(shù)據(jù)包含version字段。第二步進行更新操作，如果此時記錄已經(jīng)被修改則version字段已經(jīng)發(fā)生變化，無法更新成功：
update table set xxx,
version = #{version} + 1
where id = #{id}
and version = #{version}
4 文章總結(jié)
本文首先分析了為什么重試這個問題，因為對于RPC交互無響應(yīng)場景，重試策略是一種重要選擇。然后分析了DUBBO提供的六種集群容錯策略，F(xiàn)ailover作為默認策略提供了重試機制，在業(yè)務(wù)代碼沒有顯示重試情況下，仍有可能發(fā)起多次調(diào)用，這必須引起重視。最后我們分析了幾種常用冪等方案，希望本文對大家有所幫助。
JAVA前線
歡迎大家關(guān)注公眾號「JAVA前線」查看更多精彩分享，主要包括源碼分析、實際應(yīng)用、架構(gòu)思維、職場分享、產(chǎn)品思考等等，同時也非常歡迎大家加我微信「java_front」一起交流學(xué)習(xí)