java重試機制使用RPC必須考慮冪等性原理解析
文章概述
在RPC場景中因為重試或者沒有實現(xiàn)冪等機制而導致的重復數(shù)據(jù)問題,必須引起大家重視,有可能會造成例如一次購買創(chuàng)建多筆訂單,一條通知信息被發(fā)送多次等問題,這是技術人員必須面對和解決的問題。
有人可能會說:當調(diào)用失敗時程序并沒有顯示重試,為什么還會產(chǎn)生重復數(shù)據(jù)問題呢?這是因為即使沒有顯示重試,RPC框架在集群容錯機制中自動進行了重試,這個問題必須引起關注。
本文我們以DUBBO框架為例分析為什么重試,怎么做重試,怎么做冪等三個問題。
1 為什么重試
如果簡單對一個RPC交互過程進行分類,我們可以分為三類:響應成功、響應失敗、沒有響應。
對于響應成功和響應失敗這兩種情況,消費者很好處理。因為響應信息明確,所以只要根據(jù)響應信息,繼續(xù)處理成功或者失敗邏輯即可。但是沒有響應這種場景比較難處理,這是因為沒有響應可能包含以下情況:
(1) 生產(chǎn)者根本沒有接收到請求
(2) 生產(chǎn)者接收到請求并且已處理成功,但是消費者沒有接收到響應
(3) 生產(chǎn)者接收到請求并且已處理失敗,但是消費者沒有接收到響應
假設你是一名RPC框架設計者,究竟是選擇重試還是放棄調(diào)用呢?其實最終如何選擇取決于業(yè)務特性,有的業(yè)務本身就具有冪等性,但是有的業(yè)務不能允許重試否則會造成重復數(shù)據(jù)。
那么誰對業(yè)務特性最熟悉呢?答案是消費者,因為消費者作為調(diào)用方肯定最熟悉自身業(yè)務,所以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
快速失敗策略。消費者只消費一次服務,當發(fā)生異常時則直接拋出
(3) Failsafe
安全失敗策略。消費者只消費一次服務,如果消費失敗則包裝一個空結(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è)務代碼沒有顯示重試,也有可能多次執(zhí)行消費邏輯從而造成重復數(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 { // 服務消費發(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è)務異常直接拋出 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é)果返回給消費者,那么再次重試可能就會造成重復數(shù)據(jù)問題。
2.2.2 Failfast
快速失敗策略。消費者只消費一次服務,當發(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 { // 服務消費發(fā)起遠程調(diào)用 return invoker.invoke(invocation); } catch (Throwable e) { // 服務消費失敗不重試直接拋出異常 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
安全失敗策略。消費者只消費一次服務,如果消費失敗則包裝一個空結(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); // 服務消費發(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)造定時任務 RetryTimerTask retryTimerTask = new RetryTimerTask(loadbalance, invocation, invokers, lastInvoker, retries, RETRY_FAILED_PERIOD); try { // 定時任務放入定時器等待執(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); // 消費服務發(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); // 如果服務消費失敗則記錄失敗請求 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; // 服務消費發(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); // 防止重復增加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ù)重復問題,那么在使用中必須考慮冪等性。冪等性是指一次操作與多次操作產(chǎn)生結(jié)果相同,并不會因為多次操作而產(chǎn)生不一致性。常見冪等方案有取消重試、冪等表、數(shù)據(jù)庫鎖、狀態(tài)機。
3.1 取消重試
取消重試有兩種方法,第一是設置重試次數(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 冪等表
假設用戶支付成功后,支付系統(tǒng)將支付成功消息,發(fā)送至消息隊列。物流系統(tǒng)訂閱到這個消息,準備為這筆訂單創(chuàng)建物流單。
但是消息隊列可能會重復推送,物流系統(tǒng)有可能接收到多次這條消息。我們希望達到效果是:無論接收到多少條重復消息,只能創(chuàng)建一筆物流單。
解決方案是冪等表方案。新建一張冪等表,該表就是用來做冪等,無其它業(yè)務意義,有一個字段名為key建有唯一索引,這個字段是冪等標準。
物流系統(tǒng)訂閱到消息后,首先嘗試插入冪等表,訂單編號作為key字段。如果成功則繼續(xù)創(chuàng)建物流單,如果訂單編號已經(jīng)存在則違反唯一性原則,無法插入成功,說明已經(jīng)進行過業(yè)務處理,丟棄消息。
這張表數(shù)據(jù)量會比較大,我們可以通過定時任務對數(shù)據(jù)進行歸檔,例如只保留7天數(shù)據(jù),其它數(shù)據(jù)存入歸檔表。
還有一種廣義冪等表就是我們可以用Redis替代數(shù)據(jù)庫,在創(chuàng)建物流單之前,我們可以檢查Redis是否存在該訂單編號數(shù)據(jù),同時可以為這類數(shù)據(jù)設置7天過期時間。
3.3 狀態(tài)機
物流單創(chuàng)建成功后會發(fā)送消息,訂單系統(tǒng)訂閱到消息后更新狀態(tài)為完成,假設變更是將訂單狀態(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交互無響應場景,重試是一種重要選擇。然后分析了DUBBO提供的六種集群容錯策略,F(xiàn)ailover作為默認策略提供了重試機制,在業(yè)務代碼沒有顯示重試情況下,仍有可能發(fā)起多次調(diào)用,這必須引起重視。
最后我們分析了幾種常用冪等方案,希望本文對大家有所幫助,更多關于重試機制RPC冪等性的資料請關注腳本之家其它相關文章!
相關文章
RocketMQ之NameServer架構(gòu)設計及啟動關閉流程源碼分析
這篇文章主要為大家介紹了RocketMQ之NameServer架構(gòu)設計及啟動關閉流程源碼分析詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步2021-11-11解決SpringBoot項目啟動后網(wǎng)頁顯示Please sign in的問題
這篇文章主要介紹了解決SpringBoot項目啟動后網(wǎng)頁顯示Please sign in的問題,本文給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2021-04-04JAVA不可變類(immutable)機制與String的不可變性(推薦)
這篇文章主要介紹了JAVA不可變類(immutable)機制與String的不可變性(推薦)的相關資料,非常不錯,具有參考借鑒價值,需要的朋友可以參考下2016-08-08SpringMVC攔截器——實現(xiàn)登錄驗證攔截器的示例代碼
本篇文章主要介紹了SpringMVC攔截器——實現(xiàn)登錄驗證攔截器的示例代碼,具有一定的參考價值,感興趣的小伙伴們可以參考一下。2017-02-02