2. 程式人生 > >分散式事務解決方案之訊息最終一致性（可靠訊息服務）下篇

分散式事務解決方案之訊息最終一致性（可靠訊息服務）下篇

阿新 • • 發佈：2018-12-31
背景：1.支付成功 通知訂單完成2.訂單完成，通知會計記賬上游訂單服務，必須開放可查詢訂單狀態介面，判斷訊息是否可以傳送下游會計消費成功後，必須回撥訊息服務，ACK操作（約束：冪等性。 例如：訊息id等）流程：訂單服務： 預儲存訊息 -> 訂單完成 -> 確認傳送訊息 會計服務：消費訂單訊息 -> 完成記賬 -> 確認訊息已消費訊息生命週期： 預存儲 -> 傳送中 -> 銷燬/確認消費1.Create MySql Message Table
ROP TABLE IF EXISTS `rp_transaction_message`;
CREATE TABLE `rp_transaction_message` (
  `id` varchar(50) NOT NULL DEFAULT '' COMMENT '主鍵ID',
  `version` int(11) NOT NULL DEFAULT '0' COMMENT '版本號',
  `editor` varchar(100) DEFAULT NULL COMMENT '修改者',
  `creater` varchar(100) DEFAULT NULL COMMENT '建立者',
  `edit_time` datetime DEFAULT NULL COMMENT '最後修改時間',
  `create_time` datetime NOT NULL DEFAULT '0000-00-00 00:00:00' COMMENT '建立時間',
  `message_id` varchar(50) NOT NULL DEFAULT '' COMMENT '訊息ID',
  `message_body` longtext NOT NULL COMMENT '訊息內容',
  `message_data_type` varchar(50) DEFAULT NULL COMMENT '訊息資料型別',
  `consumer_queue` varchar(100) NOT NULL DEFAULT '' COMMENT '消費佇列',
  `message_send_times` smallint(6) NOT NULL DEFAULT '0' COMMENT '訊息重發次數',
  `areadly_dead` varchar(20) NOT NULL DEFAULT '' COMMENT '是否死亡',
  `status` varchar(20) NOT NULL DEFAULT '' COMMENT '狀態',
  `remark` varchar(200) DEFAULT NULL COMMENT '備註',
  `field` varchar(1024) DEFAULT NULL COMMENT '擴充套件欄位',
  PRIMARY KEY (`id`),
  KEY `AK_Key_2` (`message_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
2.Message API （訊息服務API）

不需要配置MQ自身重發機制，一定時間內不成功 ，規定次數內重發，重發時間可設定梯度

public interface RpTransactionMessageService {

	/**
	 * 預儲存訊息. 
	 */
	public int saveMessageWaitingConfirm(RpTransactionMessage rpTransactionMessage) throws MessageBizException;
	
	
	/**
	 * 確認併發送訊息.
	 */
	public void confirmAndSendMessage(String messageId) throws MessageBizException;

	
	/**
	 * 儲存併發送訊息.
	 */
	public int saveAndSendMessage(RpTransactionMessage rpTransactionMessage) throws MessageBizException;

	
	/**
	 * 直接傳送訊息.
	 */
	public void directSendMessage(RpTransactionMessage rpTransactionMessage) throws MessageBizException;
	
	
	/**
	 * 重發訊息.
	 */
	public void reSendMessage(RpTransactionMessage rpTransactionMessage) throws MessageBizException;
	
	
	/**
	 * 根據messageId重發某條訊息.
	 */
	public void reSendMessageByMessageId(String messageId) throws MessageBizException;
	
	
	/**
	 * 將訊息標記為死亡訊息.
	 */
	public void setMessageToAreadlyDead(String messageId) throws MessageBizException;


	/**
	 * 根據訊息ID獲取訊息
	 */
	public RpTransactionMessage getMessageByMessageId(String messageId) throws MessageBizException;

	/**
	 * 根據訊息ID刪除訊息
	 */
	public void deleteMessageByMessageId(String messageId) throws MessageBizException;
	
	
	/**
	 * 重發某個訊息佇列中的全部已死亡的訊息.
	 */
	public void reSendAllDeadMessageByQueueName(String queueName, int batchSize) throws MessageBizException;
	
	/**
	 * 獲取分頁資料
	 */
	PageBean listPage(PageParam pageParam, Map<String, Object> paramMap) throws MessageBizException;


}
3.Message APP （訊息確認，訊息恢復）
/**
 * 訊息定時器介面
 */
public interface MessageScheduled {

	/**
	 * 處理狀態為“待確認”但已超時的訊息.
	 */
	public void handleWaitingConfirmTimeOutMessages();

	/**
	 * 處理狀態為“傳送中”但超時沒有被成功消費確認的訊息
	 */
	public void handleSendingTimeOutMessage();

}
/**
 * message業務處理類
 */
@Component("messageBiz")
public class MessageBiz {

	private static final Log log = LogFactory.getLog(MessageBiz.class);

	@Autowired
	private RpTradePaymentQueryService rpTradePaymentQueryService;
	@Autowired
	private RpTransactionMessageService rpTransactionMessageService;

	/**
	 * 處理[waiting_confirm]狀態的訊息
	 * 
	 * @param messageMap
	 */
	public void handleWaitingConfirmTimeOutMessages(Map<String, RpTransactionMessage> messageMap) {
		log.debug("開始處理[waiting_confirm]狀態的訊息,總條數[" + messageMap.size() + "]");
		// 單條訊息處理（目前該狀態的訊息，消費佇列全部是accounting，如果後期有業務擴充，需做佇列判斷，做對應的業務處理。）
		for (Map.Entry<String, RpTransactionMessage> entry : messageMap.entrySet()) {
			RpTransactionMessage message = entry.getValue();
			try {
				log.debug("結束處理[waiting_confirm]訊息ID為[" + message.getMessageId() + "]的訊息");
			} catch (Exception e) {
				log.error("處理[waiting_confirm]訊息ID為[" + message.getMessageId() + "]的訊息異常：", e);
			}
		}
	}

	/**
	 * 處理[SENDING]狀態的訊息
	 * 
	 * @param messageMap
	 */
	public void handleSendingTimeOutMessage(Map<String, RpTransactionMessage> messageMap) {
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		log.debug("開始處理[SENDING]狀態的訊息,總條數[" + messageMap.size() + "]");

		// 根據配置獲取通知間隔時間
		Map<Integer, Integer> notifyParam = getSendTime();

		// 單條訊息處理
		for (Map.Entry<String, RpTransactionMessage> entry : messageMap.entrySet()) {
			RpTransactionMessage message = entry.getValue();
			try {
				log.debug("開始處理[SENDING]訊息ID為[" + message.getMessageId() + "]的訊息");
				// 判斷髮送次數
				int maxTimes = Integer.valueOf(PublicConfigUtil.readConfig("message.max.send.times"));
				log.debug("[SENDING]訊息ID為[" + message.getMessageId() + "]的訊息,已經重新發送的次數[" + message.getMessageSendTimes() + "]");

				// 如果超過最大發送次數直接退出
				if (maxTimes < message.getMessageSendTimes()) {
					// 標記為死亡
					rpTransactionMessageService.setMessageToAreadlyDead(message.getMessageId());
					continue;
				}
				// 判斷是否達到傳送訊息的時間間隔條件
				int reSendTimes = message.getMessageSendTimes();
				int times = notifyParam.get(reSendTimes == 0 ? 1 : reSendTimes);
				long currentTimeInMillis = Calendar.getInstance().getTimeInMillis();
				long needTime = currentTimeInMillis - times * 60 * 1000;
				long hasTime = message.getEditTime().getTime();
				// 判斷是否達到了可以再次傳送的時間條件
				if (hasTime > needTime) {
					log.debug("currentTime[" + sdf.format(new Date()) + "],[SENDING]訊息上次傳送時間[" + sdf.format(message.getEditTime()) + "],必須過了[" + times + "]分鐘才可以再發送。");
					continue;
				}

				// 重新發送訊息
				rpTransactionMessageService.reSendMessage(message);

				log.debug("結束處理[SENDING]訊息ID為[" + message.getMessageId() + "]的訊息");
			} catch (Exception e) {
				log.error("處理[SENDING]訊息ID為[" + message.getMessageId() + "]的訊息異常：", e);
			}
		}

	}

	/**
	 * 根據配置獲取通知間隔時間
	 * 
	 * @return
	 */
	private Map<Integer, Integer> getSendTime() {
		//TODO...   config 配置 次數相對應的時間間隔
	}
}
4.Message QUEUE（訊息消費）
·消費訊息，完成業務操作後，呼叫訊息服務ACK確認訊息已經被消費5.Message WEB （訊息管理）訊息管理視覺化介面，處理死亡訊息等優化1.訂單完成後，呼叫訊息RPC確認訊息可能超時，訂單將會被回滾，但是訊息已經被確認，導致會計成功記賬方案：確認訊息設定成非同步<dubbo:reference interface="com.roncoo.pay.service.message.api.RpTransactionMessageService" id="rpTransactionMessageService" check="false"><dubbo:method name="confirmAndSendMessage" async="true" return="false" /></dubbo:reference>2.訊息儲存DB方案：redis，mongodb3.被動方業務冪等性判斷方案：業務操作成功，記錄訊息id4.服務部署叢集導致任務重複排程方案：分散式任務排程（噹噹開源elastic-job，分散式作業排程框架）5.實時訊息服務方案：rabbitMQ，rocketMQ6.效能提高，解耦方案：每個業務使用一套訊息服務

相關推薦

分散式事務解決方案之訊息最終一致性（可靠訊息服務）下篇

背景：1.支付成功 通知訂單完成2.訂單完成，通知會計記賬上游訂單服務，必須開放可查詢訂單狀態介面，判斷訊息是否可以傳送下游會計消費成功後，必須回撥訊息服務，ACK操作（約束：冪等性。 例如：訊息id等）流程：訂單服務： 預儲存訊息 -> 訂單完成 ->

曹工雜談：分散式事務解決方案之基於本地訊息表實現最終一致性

# 曹工雜談：分散式事務解決方案之基於本地訊息表實現最終一致性 # 前言 為什麼寫這個？其實我這邊的業務場景，嚴格來說，不算是典型的分散式事務，需求是這樣說的：因為我這邊負責的一個服務消費者consumer，是使用者登入的入口；正常情況下，登入時候要走使用者中心，這是個單獨的服務；如果使用者中心掛了，我這

分散式事務解決方案(二)【基於可靠訊息的最終一致性】

2. 最終一致性(基於可靠訊息) 2.1 訊息傳送的一致性 指產生訊息的業務動作與訊息傳送的一致。（也就是說，如果業務操作成功，那麼由這個業務操作所產生的訊息一定要成功投遞出去，否則就丟訊息） 2.1.1 如何保障訊息傳送一致性 處理方式1

[轉載]使用訊息佇列實現分散式事務-公認較為理想的分散式事務解決方案

前陣子從支付寶轉賬1萬塊錢到餘額寶，這是日常生活的一件普通小事，但作為網際網路研發人員的職業病，我就思考支付寶扣除1萬之後，如果系統掛掉怎麼辦，這時餘額寶賬戶並沒有增加1萬，資料就會出現不一致狀況了。 上述場景在各個型別的系統中都能找到相似影子，比如在電商系統中，當有使用者下單後，除了在訂單表插

使用訊息佇列實現分散式事務-公認較為理想的分散式事務解決方案（三）

  Begin transaction update A set amount=amount-10000 where userId=1; update B set amount=amount+10000 where userId=1; End t

億級流量架構之分散式事務解決方案對比

上一篇文章( [億級流量架構之分散式事務思路及方法](https://www.cnblogs.com/Courage129/p/14433462.html))中梳理事務到分散式事務的演變過程, 以及分散式事務的處理思路,這篇文章主要從應用的角度對比目前較為流行的一些分散式事務方案,以及一些商業應用。 想讓資

分散式事務解決方案---------LCN

1.過多的原理我就不一一介紹了，我就用一個例項來展示LCN分散式事務解決方案的應用。 tx-lcn https://gitee.com/wangliang1991/tx-lcn springcloud-demo版本的demo https://github.com/codingapi/

聊聊微服務架構及分散式事務解決方案！

分散式事務場景如何設計系統架構及解決資料一致性問題，個人理解最終方案把握以下原則就可以了，那就是：大事務=小事務（原子事務）+非同步（訊息通知），解決分散式事務的最好辦法其實就是不考慮分散式事務，將一個大的業務進行拆分，整個大的業務流程，轉化成若干個小的業務流程，然後通過設計補償流程從而考慮最終一致性。什麼是

Spring Cloud分散式事務解決方案

開源專案 我們利用訊息佇列實現了分散式事務的最終一致性解決方案，請大家圍觀。可以參考Github CoolMQ原始碼，專案支援網站: http://rabbitmq.org.cn，最新文章或實現會更新在上面 二 前言 阿里2017雲棲大會《破解世界性技術難題！GTS

更多免費初級中級高階大資料java視訊教程下載 加(微***信((號keepper,請備註java或掃下面2二3維4碼第31： 2017年7月最新微服務架構的分散式事務解決方案價值1399

更多免費初級中級高階大資料java視訊教程下載 加(微***信((號keepper,請備註java或掃下面2二3維4碼第31： 2017年7月最新微服務架構的分散式事務解決方案價值1399java視訊教程01 課程介紹.wmvjava視訊教程02 解決方案的效果演示（結合支付系統真實應用場景）.mp4java

分散式事務解決方案

背景 本地事務 一個單體應用中事務由一個數據庫管理,並且限制在單個程序中的事務. 不涉及多個數據來源. 優點:支援嚴格的ACID,可靠,高效,簡單. 不足:沒有分散式處理能力 隔離的最小單位有資源管理器(資料庫)決定,如果資料庫

2019最新微服務架構的分散式事務解決方案課程 共31課

教程內容：微服務倡導將複雜的單體應用拆分為若干個功能簡單、鬆耦合的服務，這樣可以降低開發難度、增強擴充套件性、便於敏捷開發。當前被越來越多的開發者推崇，很多網際網路行業巨頭、開源社群等都開始了微服務的討論和實踐。Hailo有160個不同服務構成，NetFlix有大約600個服務。國內方面，阿里巴巴、

私有云落地解決方案之openstack高可用（pike版本）-訊息佇列

作者：【吳業亮】 1、安裝軟體包 yum install erlang rabbitmq-server –y 2、啟動服務並設定開機啟動 systemctl enable rabbitm

java微服務架構的分散式事務解決方案

分散式系統架構中，分散式事務問題是一個繞不過去的挑戰。而微服務架構的流行，讓分散式事問題日益突出！ 下面我們以電商購物支付流程中，在各大參與者系統中可能會遇到分散式事務問題的場景進行詳細的分析！   如上圖所示，假設三大參與平臺（電商平臺、支付平臺、銀行）的系統都

阿里微服務架構下分散式事務解決方案－GTS

雖然微服務現在如火如荼，但對其實踐其實仍處於初級階段。即使網際網路巨頭的實踐也大多是試驗層面，鮮有核心業務系統微服務化的案例。GTS是目前業界第一款，也是唯一的一款通用的解決微服務分散式事務問題的中介軟體，而且可以保證資料的強一致性。本文將對GTS做出深入解讀。 微服務倡導將複雜的單體應用拆分為若干個功能簡

微服務架構的分散式事務解決方案

分散式系統架構中，分散式事務問題是一個繞不過去的挑戰。而微服務架構的流行，讓分散式事問題日益突出！ 下面我們以電商購物支付流程中，在各大參與者系統中可能會遇到分散式事務問題的場景進行詳細的分析！ 如上圖所示，假設三大參與平臺（電商平臺、支付平臺、銀行）的系統都做了

阿里巴巴開源分散式事務解決方案 Fescar

Fescar 是 阿里巴巴 開源的 分散式事務中介軟體，以 高效 並且對業務 0 侵入 的方式，解決 微服務 場景下面臨的分散式事務問題。 1. 什麼是微服務化帶來的分散式事務問題？ 首先，設想一個傳統

分散式事務解決方案框架（LCN）

事物概念 事物特性(ACID) 原子性（A） 所謂的原子性就是說，在整個事務中的所有操作，要麼全部完成，要麼全部不做，沒有中間狀態。對於事務在執行中發生錯誤，所有的操作都會被回滾，整個事務就像從沒被執行過一樣。 一致性（C） 事務的執行必須保證系統的一致性，就拿轉賬

.NET Core 事件匯流排,分散式事務解決方案：CAP

背景 相信前面幾篇關於微服務的文章也介紹了那麼多了，在構建微服務的過程中確實需要這麼一個東西，即便不是在構建微服務，那麼在構建分散式應用的過程中也會遇到分散式事務的問題，那麼 CAP 就是在這樣的背景下誕生的。 最初打算做這個東西是在去年（2016）年底，最初是為了解決分散式系統中的分散式事務的問題，然後當時

微服務架構下分散式事務解決方案 —— 阿里GTS

原文地址:https://yq.aliyun.com/articles/5420201 微服務的發展微服務倡導將複雜的單體應用拆分為若干個功能簡單、鬆耦合的服務，這樣可以降低開發難度、增強擴充套件性、便於敏捷開發。當前被越來越多的開發者推崇，很多網際網路行業巨頭、開源社群等都