通過RabitMQ實作分布式事務題
- 前言
- 業務需求
- 核心原理
- 核心難點
- 解決問題思路
- 代碼實作
- 其他
- 總結
前言
這篇文章是通過學習嗶哩嗶哩中的視頻“阿里架構師如何30分鐘基于MQ解決分布式事務問題”，原視頻連接https://www.bilibili.com/video/BV15p4y1D7d7?p=2
本人水平有限，如有誤導，歡迎斧正，一起學習，共同進步！
業務需求
這是一次簡單的模擬分布式事務的：假設我們現在需要支付寶向余額寶轉賬100元，支付寶(系統A)向余額寶(系統B)轉賬100元，支付寶余額-100，余額寶余額加100，這涉及到兩個系統，需要用到分布式事務，
核心原理
系統A(支付寶系統)完成相關業務(此次是扣款操作)后，向MQ中發送訊息，同時往系統A(支付寶系統)的訊息表中插入一條訊息，狀態是未確認(為什么要往訊息表插入資料后面會有解釋)；系統B(余額寶系統)從MQ中拿到訊息，完成相應的業務(此次是加款操作)后，往系統B(余額寶系統)的訊息表插入一條訊息，并通知系統A(支付寶系統)，當前訊息已經被成功操作(我這加款成功啦，你可以放心啦)，系統A(支付寶系統)監聽到系統B(余額寶系統)發送來的訊息后，修改系統A(支付寶系統)本地的訊息表的這條訊息的狀態,改為已確認；利用定時任務去定時的掃描系統A的訊息表，若是有狀態是未確認的訊息，則再次將該訊息發送到MQ中，等待訊息消費者的消費，
核心難點
1、訊息丟失：假設訊息生產者(此處是系統A，支付寶系統)向MQ發送訊息正常發送；訊息消費者(此處是系統B，余額寶系統)正常監聽到了MQ中的訊息，但是由于某種原因，訊息消費者未執行相應的業務邏輯(此處就是余額寶未成功+100，說極端點，比如我余額寶資料庫直接宕機了，雖然可能性比較小，但是要考慮到這種情況)，但是MQ認為訊息消費者已經正常的消費了訊息，就把訊息洗掉了，這就是典型的訊息丟失問題，
2、重復消費訊息(保證訊息的冪等性)：假設訊息生產者(系統A，支付寶系統)向MQ發送訊息成功，訊息消費者(系統B，余額寶系統)成功監聽到了訊息并成功消費(就是說余額寶成功+100了)，但是訊息消費者向訊息生產者發送通知的時候，由于網路原因或其他原因，發送失敗(余額寶向支付寶說，我錢增加啦，你放心吧，這個步驟發送失敗了，這一步也是通過MQ來實作通信的)，此時，由于系統A遲遲未收到系統B的確認訊息，那么系統A的本地訊息表的狀態就一直是未確認，定時器定時掃描的時候，就會把這條訊息繼續發送到MQ中，這就會導致支付寶扣了一次款(-100)，余額寶增加了兩次款(+100，第二次又+100)，是不是做夢都要笑醒了~
解決問題思路
1、解決第一個問題，就是在系統A中，增加一個訊息表(這個訊息表是系統A的)，只要我相關業務執行成功了(支付寶-100成功了)，那么就往這個訊息表增加一條資料(訊息表核心欄位有ID，狀態，這個ID類似于Dubbo的服務追蹤的穿透ID，流水ID，是倆系統溝通的憑證，有點類似于token一樣的，因為我需要通過這個ID來定位到某一個訊息)，只有我監聽到系統B給我的反饋了以后(系統B給訊息A發送訊息的時候帶著這個訊息的ID，可以定位到這個訊息)，我才將這個訊息的狀態改為“已消費”，否則就是默認的未消費，定時器會定時的掃描這個表，將狀態是未消費的訊息，重新發送到MQ中，
2、解決第二個問題，就是在系統B中，也增加一個訊息表(這個訊息表是系統B自己的)，只要我成功的執行完了我的業務代碼(余額寶+100成功)以后，我就在這個訊息表增加一個訊息，即使某種原因，系統A未收到系統B的確認訊息，無非就是把同樣的訊息在重新發送到MQ中，但是系統B本地有已經消費過的訊息存根，系統B監聽到訊息以后，先去系統B的訊息表(訊息本地存根)中根據監聽到的訊息ID查，若是能查到，則證明此條訊息已經被消費過了，則不在消費，直接給系統A發送訊息，說我成功消費了，你改這個訊息的狀態吧，別在給我發這條訊息了，
代碼實作
我個人的github鏈接(僅供參考): https://github.com/zhengtianliang/distributed_transaction_rabbitmq.
1、系統A(支付寶系統)的核心業務代碼
package com.alipay.service.impl;
import com.alipay.dao.AccountMapper;
import com.alipay.dao.AlipayMessageMapper;
import com.alipay.entity.Account;
import com.alipay.entity.AlipayMessage;
import com.alipay.entity.MessageStatus;
import com.alipay.rabbitmq.RabbitMqSender;
import com.alipay.service.OrderService;
import com.alipay.util.json.JsonUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.UUID;
/**
* @author ZhengTianLiang
* @date 2020/9/12 22:00
* @desc service
*/
@Service
public class OrderServiceImpl implements OrderService {
@Resource
private AccountMapper accountMapper;
@Resource
private AlipayMessageMapper alipayMessageMapper;
@Resource
private TransactionTemplate transactionTemplate;
@Resource
private RabbitMqSender rabbitMqSender;
/**
* @author ZhengTianLiang
* @date 2020/9/12 21:57
* @desc 更新余額操作 系統1(支付寶系統的)的扣款操作
*/
// @Transactional /**
// * 不能使用這個注解的原因是，mq發送訊息是比較占用資源的，但是這個又不是必須成功的，也就是說步驟1、2
// * 是必須同步的，要么同事成功，要么同事失敗，而步驟3是可以失敗的，因為我要是沒往ma發送成功訊息，無非就是訊息存根表
// * 的狀態一直是未確認，那我定時任務會再次發送的，但是步驟1和步驟2是必須要么同時成功，要么同時失敗，
// * 所以我們要盡量避免使用宣告式事務，可以使用編程式事務(因為往mq發送訊息比較消耗記憶體，又不是事務,無須與前倆系結的)
// */
@Override
public void updateAmount(int amount, String userId) {
// 創建一個訊息存根物件，用來插入到訊息存根表、mq發送訊息也用這個物件
AlipayMessage message = new AlipayMessage(UUID.randomUUID().toString(), userId, LocalDateTime.now(), MessageStatus.NOCONFIRM);
// 使用編程式事務
Integer execute = transactionTemplate.execute(new TransactionCallback<Integer>() {
@Override
public Integer doInTransaction(TransactionStatus transactionStatus) {
// 1、account進行扣款操作
Account account = new Account(userId, amount, LocalDateTime.now());
int i = accountMapper.updateAmountById(account);
// 2、第一步執行成功的話，往訊息存根表中插入訊息
if (i == 1) {
return alipayMessageMapper.insertMessage(message); // 訊息存根表成功插入了資料
}
return 0; // 未成功插入資料
}
});
if (execute > 0){ // 3、步驟1,2都成功的話，往mq中發送訊息
rabbitMqSender.sendMessage("","",message);
}
}
/**
* @author ZhengTianLiang
* @date 2020/9/13 10:19
* @desc 支付寶監聽到了余額寶發來的訊息以后，修改訊息存根表的狀態，改為“已確認”
*/
/**
* 有些童鞋可能要問了，這明明是一個單表操作，為啥要加上@Transactional呢？
* 因為如果不加事務的話，整個連接池就會有兩個鏈接，也就是說，在同一個類中，有的是用了@Transactional注解，
* 有的類沒有用@Transactional注解，那么就會導致，有兩個連接池物件，一個是spring維護的，一個是mybatis維護的
* 假設沒有用@Transactional注解，又用到了mybatis的update，那么mybatis的原始碼就會從連接池中獲取一個鏈接物件，而
* 這個鏈接物件是由mybatis維護、管理的，
* 而如果使用了@Transactional注解(或者編程式事務)，即使用到了mybatis的update方法，那么也是從spring中拿的連接池物件，
* 這一切都是由spring維護、管理的，
* 總結：下面這個方法，加不加@Transactional注解，執行的效果是一樣的，但是加了的話，連接池物件會從2個變成1個，
* 降低了整個系統的開銷，本來設計的初衷就是為了提高系統的吞吐量，要是有兩個連接池物件，與設計初衷不符，所以加了此注解
*/
@Transactional
@Override
public void updateMessageStatus(String message) { // 余額寶傳過來的訊息的狀態是confirm已確認的
AlipayMessage alipayMessage = JsonUtils.jsonToBean(message, AlipayMessage.class);
alipayMessageMapper.updateMessageSatus(alipayMessage);
}
}
上面這塊代碼有2點需要注意的，第一點是最好不要使用宣告式事務@Transactional，而用編程式事務替代它，因為這三個步驟中(第一步扣款、第二步訊息存根表插入資料、第三步往mq中發送訊息)，前兩步是不許一起的，而第三步并非是與前兩步系結一起的(前兩步必須同時成功或失敗，但是第三步沒必要，我即使失敗了，沒往ma發送成功訊息，無非就是訊息存根表的狀態一直是未確認，那我定時任務會再次發送的，但是步驟1和步驟2是必須要么同時成功，要么同時失敗)，第二點則是spring管理的連接池物件和mybatis管理的連接池物件的區別，解釋在下面：，
/**
* 有些童鞋可能要問了，這明明是一個單表操作，為啥要加上@Transactional呢？
* 因為如果不加事務的話，整個連接池就會有兩個鏈接，也就是說，在同一個類中，有的是用了@Transactional注解，
* 有的類沒有用@Transactional注解，那么就會導致，有兩個連接池物件，一個是spring維護的，一個是mybatis維護的
* 假設沒有用@Transactional注解，又用到了mybatis的update，那么mybatis的原始碼就會從連接池中獲取一個鏈接物件，而
* 這個鏈接物件是由mybatis維護、管理的，
* 而如果使用了@Transactional注解(或者編程式事務)，即使用到了mybatis的update方法，那么也是從spring中拿的連接池物件，
* 這一切都是由spring維護、管理的，
* 總結：下面這個方法，加不加@Transactional注解，執行的效果是一樣的，但是加了的話，連接池物件會從2個變成1個，
* 降低了整個系統的開銷，本來設計的初衷就是為了提高系統的吞吐量，要是有兩個連接池物件，與設計初衷不符，所以加了此注解
*/
2、RabbitMQ的相關配置
- a.RabbitMqConfig
package com.alipay.rabbitmq;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author ZhengTianLiang
* @date 2020/9/12 22:27
* @desc rabbitmq的配置類
*/
@Configuration
public class RabbitMqConfig {
/**
* rabbitmq中，訊息發送者只需要知道交換機的名稱就行(還需攜帶路由鍵)，不需要知道佇列的名稱
* rabbitmq中，訊息消費者只需要知道佇列名稱就行，不需要知道交換器和路由鍵的名稱
*/
/**
* @author ZhengTianLiang
* @date 2020/9/12 22:42
* @desc 往rabbitmq的broker里面創建一個佇列
*/
@Bean(name = "message")
public Queue getQueue() {
return new Queue("zheng.alipay.message");
}
/**
* @author ZhengTianLiang
* @date 2020/9/12 22:46
* @desc 創建交換器
*/
@Bean
public TopicExchange getExchange(){
return new TopicExchange("zheng.alipay.exchange");
}
@Bean
Binding bindingExchangeMessage(@Qualifier("message") Queue getQueue,TopicExchange getExchange){
return BindingBuilder.bind(getQueue).to(getExchange()).with("zheng.alipay.routkey");
}
}
-b.RabbitMqSender
package com.alipay.rabbitmq;
import com.alipay.entity.AlipayMessage;
import com.alipay.util.json.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* @author ZhengTianLiang
* @date 2020/9/12 22:29
* @desc 訊息發送者
*/
@Slf4j
@Component
public class RabbitMqSender {
@Resource
private AmqpTemplate amqpTemplate;
/**
* @author ZhengTianLiang
* @date 2020/9/12 22:29
* @desc 從支付寶系統發送訊息到mq中
* @param exchange 交換機名稱
* @param routingKey 路由鍵名稱
* @param message 要發送的訊息
*/
public void sendMessage(String exchange, String routingKey, AlipayMessage message) {
log.info("支付寶系統往mq中發送了訊息：" + message.getMessageId());
amqpTemplate.convertAndSend(exchange, routingKey, JsonUtils.toJson(message));
}
}
- c.MessageListener
package com.alipay.rabbitmq.listener;
import com.alipay.entity.AlipayMessage;
import com.alipay.service.OrderService;
import com.alipay.util.json.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* @author ZhengTianLiang
* @date 2020/9/13 12:08
* @desc 支付寶系統的mq的監聽器,
*/
@Slf4j
@Component
public class MessageListener {
@Resource
private OrderService orderService;
/**
* @author ZhengTianLiang
* @date 2020/9/13 12:09
* @desc 監聽余額寶發送的訊息，用來改變支付寶自己的訊息存根表的訊息狀態
*/
@RabbitListener(queues = "zheng.moneypay.message") // 它監聽的是余額寶的訊息佇列
public void process(String message){
orderService.updateMessageStatus(message); // 更新訊息的狀態
}
}
3、系統B(余額寶系統)的核心業務邏輯(是寫在mq的監聽器里面了)
package com.rabbitmq.listener;
import com.entity.AlipayMessage;
import com.entity.MessageStatus;
import com.entity.MoneypayMessage;
import com.rabbitmq.RabbitMqSender;
import com.service.OrderService;
import com.util.json.JsonUtils;
import com.util.mapper.MapperUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import javax.annotation.Resource;
/**
* @author ZhengTianLiang
* @date 2020/9/13 11:08
* @desc mq的監聽器, 也是核心的業務代碼
*/
@Slf4j
@Component
public class MessageListener {
@Resource
private OrderService orderService;
@Resource
private TransactionTemplate transactionTemplate;
@Resource
private RabbitMqSender rabbitMqSender;
@Value("${moneypay.exchange}")
private String exchange;
@Value("${moneypay.routkey}")
private String routkey;
/**
* @author ZhengTianLiang
* @date 2020/9/13 11:40
* @desc 監聽訊息發送者(支付寶系統)發送來的訊息，并作出相應的業務操作
*/
@RabbitListener(queues = "zheng.alipay.message") // 訊息發送者的佇列，就是支付寶發送到的佇列的佇列名
public void process(String jsonMessage) {
AlipayMessage alipayMessage = JsonUtils.jsonToBean(jsonMessage, AlipayMessage.class);
MoneypayMessage moneypayMessage = MapperUtils.mapperBean(alipayMessage, MoneypayMessage.class);
// 1、去余額寶自己的訊息存根中查，看看能不能根據支付寶系統傳過來的訊息id來查到資料，能查到則說明消費過了；查不到則是未消費過
Integer count = orderService.queryMessageCountById(alipayMessage.getMessageId());
Boolean exec = transactionTemplate.execute(new TransactionCallback<Boolean>() {
@Override
public Boolean doInTransaction(TransactionStatus transactionStatus) {
if (count == 0) { //此訊息未消費過
orderService.updateAmount(moneypayMessage.getAmount(), moneypayMessage.getUserId());// 加款操作
orderService.insertMessage(moneypayMessage); // 往余額寶自己的訊息本地存根中插入資料
}
return true;
}
});
if (count > 0) { // 此訊息已經被消費過了，則直接通過mq去通知支付寶系統，說這個訊息消費過了
log.info("此訊息已經被消費過了,不做任何操作");
}
if (exec){ // 去mq中發訊息，通知支付寶，說此訊息已經被消費過了
alipayMessage.setStatus(MessageStatus.CONFIRM);
rabbitMqSender.sendMessage(exchange,routkey,alipayMessage);
}
}
}
4、mq的相關配置
- a.RabbitMqConfig
package com.rabbitmq;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author ZhengTianLiang
* @date 2020/9/12 22:27
* @desc rabbitmq的配置類
*/
@Configuration
public class RabbitMqConfig {
/**
* rabbitmq中，訊息發送者只需要知道交換機的名稱就行(還需攜帶路由鍵)，不需要知道佇列的名稱
* rabbitmq中，訊息消費者只需要知道佇列名稱就行，不需要知道交換器和路由鍵的名稱
*/
/**
* @author ZhengTianLiang
* @date 2020/9/12 22:42
* @desc 往rabbitmq的broker里面創建一個佇列
*/
@Bean(name = "message")
public Queue getQueue() {
return new Queue("zheng.moneypay.message");
}
/**
* @author ZhengTianLiang
* @date 2020/9/12 22:46
* @desc 創建交換器
*/
@Bean
public TopicExchange getExchange(){
return new TopicExchange("zheng.moneypay.exchange");
}
@Bean
Binding bindingExchangeMessage(@Qualifier("message") Queue getQueue,TopicExchange getExchange){
return BindingBuilder.bind(getQueue).to(getExchange()).with("zheng.moneypay.routkey");
}
}
-b.RabbitMqSender
package com.rabbitmq;
import com.util.json.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* @author ZhengTianLiang
* @date 2020/9/12 22:29
* @desc 訊息發送者
*/
@Slf4j
@Component
public class RabbitMqSender {
@Resource
private AmqpTemplate amqpTemplate;
/**
* @param exchange 交換機名稱
* @param routingKey 路由鍵名稱
* @param message 要發送的訊息
* @author ZhengTianLiang
* @date 2020/9/12 22:29
* @desc 從支付寶系統發送訊息到mq中
*/
public void sendMessage(String exchange, String routingKey, Object message) {
log.info("支付寶系統往mq中發送了訊息：" + message);
amqpTemplate.convertAndSend(exchange, routingKey, JsonUtils.toJson(message));
}
}
5、余額寶系統的service
package com.service.impl;
import com.dao.AccountMapper;
import com.dao.MoneypayMessageMapper;
import com.entity.Account;
import com.entity.MoneypayMessage;
import com.service.OrderService;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.UUID;
/**
* @author ZhengTianLiang
* @date 2020/9/13 11:27
* @desc service
*/
@Service
public class OrderServiceImpl implements OrderService {
@Resource
MoneypayMessageMapper messageMapper;
@Resource
AccountMapper accountMapper;
/**
* @author ZhengTianLiang
* @date 2020/9/13 11:31
* @desc 這個是賬戶的新增操作
*/
@Override
public void updateAmount(Integer amount, String userId) {
Account account = new Account(userId,amount, LocalDateTime.now());
accountMapper.updateAmountById(account);
}
/**
* @author ZhengTianLiang
* @date 2020/9/13 11:27
* @desc 根據訊息id查到訊息的數量(用來判斷該訊息是否被消費過)
*/
@Override
public Integer queryMessageCountById(String messageId) {
return messageMapper.queryCountByMessageId(messageId);
}
/**
* @author ZhengTianLiang
* @date 2020/9/13 11:27
* @desc 往余額寶系統的訊息存根表中插入資料
*/
@Override
public void insertMessage(MoneypayMessage message) {
messageMapper.insertMessage(message);
}
}
其他
- 下面是我之前學習的時候記錄的一些筆記，希望能對讀者提供一些幫助，
二、分布式事務的三種比較常見的解決方案：
1、基于XA協議的兩階段提交
①XA規范中分布式事務由AP，RM，TM三部分組成，具體點就是，應用程式(AP)，事務管理器(TM)，資源管理器(RM)，通信資源管理器(CRM)四個部分，一般，常見的事務管理器(TM)是交易中間件，常見的資源管理器(RM)是資料庫，常見的通信資源管理器(CRM)是訊息中間件
AP：定義事務邊界(定義事務開始和結束),并訪問事務邊界內的資源
RM：資源管理器，管理計算機共享的資源，許多軟體都可以去訪問這些資源，資源包含比如資料庫，檔案系統，列印機服務器等
TM：負責管理全域事務，分配事務唯一標識，監控事務的執行進度，并負責事務的提交、回滾、失敗恢復等
②兩階段協議：
第一階段：TM要求所有的RM準備提交對應的事務分支，詢問RM是否有能力保證成功的提交事務分支，RM根據自己的情況，如果判斷自己進行的作業可以被提交，那就對作業的內容就行持久化，并給TM回執OK，否則給TM回執NO，RM在發送了否定答復并回滾了已經的作業后，就可以丟棄這個事務分支資訊了，
第二階段：TM根據階段1各個RM prepare的結果，決定是提交還是回滾事務，如果所有的RM都prepare成功，那么TM通知所有的RM進行提交，如果有RM prepare回執NO的話，則TM通知所有的RM回滾自己的事務分支
③XA協議兩階段提交的優缺點：
優點：進來保持了資料的強一致性，適合對資料強一致性要求跟高的領域(并非100%保證強一致性)
缺點：實作復雜，犧牲了可用性，對性能影響比較大，不適合高并發高性能的場景
2、TCC補償機制
TCC其實就是采用的補償機制，其核心思想是，針對每個操作，都要注冊一個與其對應的確認和補償(撤銷)操作，他分為三個階段：
·Try階段，主要是對業務系統做檢測已經資源御靈
·Confirm階段主要是對業務系統做確認提交，Try階段執行成功并開始執行Confirm階段時，默認Confirm階段是不會出錯的，即：主要Try成功，Confirm一定成功
·Cancel階段主要是業務執行錯誤，需要回滾的狀態下執行的業務取消，預留資源釋放
②TCC補償機制的優缺點：
優點：相比于兩階段提交，可用性比較強(因為兩階段涉及到了鎖的概念)
缺點：資料的一致性要差一些，TCC屬于應用層的一種補償方式，所以需要程式員在實作的時候多寫很多補償的代碼，在一些場景中，一些業務流程可能用TCC不太好定義處理
3、訊息最終一致性
訊息最終一致性應該是業界使用最多的，其核心思想是將分布式事務拆分為本地事務進行處理，
①基本流程：
假設有兩個事務，第一個事務先寫入業務資料，然后在寫入訊息資料(此時會額外建立一個訊息表來記錄我要發送的訊息內容)，寫完訊息資料之后，在將資料發送給MQ(第一階段到此結束)，第二個事務要用到這個資料第二個事務就從MQ中拿到對應的資料，拿到這個訊息，在寫入業務資料，如果說事務成功了，就修改訊息表(就是事務1新建的，用來存盤要發送的訊息內容的訊息表)中的狀態，如果說事務操作這個訊息失敗了，也要給事務1說一下，然后事務1在呼叫一些補償的代碼來執行一些資料庫回滾的操作，若寫入業務事務成功，(事務1)發送到MQ失敗了，會怎么樣？則訊息表會存積大量的未處理的訊息資料，此時，會有一個另外的執行緒去定時的去掃描這個訊息表，若發現有大量的未處理的訊息，則在進行一些對應的補償邏輯、
②訊息最終一致性的優缺點：
優點：一種非常經典的實作，避免了分布式事務，實作了最終一致性
缺點：訊息表會耦合到業務系統中，若沒有封裝好的解決方案，會有很多的雜活，
總結
其實分布式的事務原理肯定都一樣：利用一個總的事務管理器來管理一個個的零散的本地事務，假如全部成功，則成功；有任意一個失敗，則回滾，所以分布式事務幾乎都是非常影響系統性能的，
現在比較常用的有TX-LCN、seata等，還有之前的基于XA兩階段提交、TCC補償機制等等，
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/37228.html
標籤：java
上一篇：大資料常見面試題之hdfs
下一篇：計算高性能[集群高性能]