一、前言

1.1 簡述

分布式事務指事務的操作位于不同的節點上，需要保證事務的 AICD 特性，
例如在下單場景下，庫存和訂單如果不在同一個節點上，就涉及分布式事務，

1.2 分布式事務的幾種解決方案

1.2.1 2PC(兩階段提交)

2PC即兩階段提交協議，是將整個事務流程分為兩個階段，準備階段（Prepare phase）、提交階段（commit phase），2是指兩個階段，P是指準備階段，C是指提交階段，
在計算機中部分關系資料庫如Oracle、MySQL支持兩階段提交協議，如下圖：

準備階段（Prepare phase）：事務管理器給每個參與者發送Prepare訊息，每個資料庫參與者在本地執行事務，并寫本地的Undo/Redo日志，此時事務沒有提交，（Undo日志是記錄修改前的資料，用于資料庫回滾，Redo日志是記錄修改后的資料，用于提交事務后寫入數據檔案）
提交階段（commit phase）：如果事務管理器收到了參與者的執行失敗或者超時訊息時，直接給每個參與者發送回滾(Rollback)訊息；否則，發送提交(Commit)訊息；參與者根據事務管理器的指令執行提交或者回滾操作，并釋放事務處理程序中使用的鎖資源，注意:必須在最后階段釋放鎖資源，

這里不重點介紹

1.2.2 XA方案

2PC的傳統方案是在資料庫層面實作的，如Oracle、MySQL都支持2PC協議，為了統一標準減少行業內不必要的對接成本，需要制定標準化的處理模型及介面標準，國際開放標準組織Open Group定義了分布式事務處理模型 DTP（Distributed Transaction Processing Reference Model），

1）在準備階段RM執行實際的業務操作，但不提交事務，資源鎖定；
2）在提交階段TM會接受RM在準備階段的執行回復，只要有任一個RM執行失敗，TM會通知所有RM執行回滾操作，否則，TM將會通知所有RM提交該事務，提交階段結束資源鎖釋放，

XA方案的問題：
1、需要本地資料庫支持XA協議，
2、資源鎖需要等到兩個階段結束才釋放，性能較差，

這里不重點介紹

1.2.3 Seata方案

Seata是由阿里中間件團隊發起的開源專案 Fescar，后更名為Seata，它是一個是開源的分布式事務框架，傳統2PC的問題在Seata中得到了解決，它通過對本地關系資料庫的分支事務的協調來驅動完成全域事務，是作業在應用層的中間件，主要優點是性能較好，且不長時間占用連接資源，它以高效并且對業務0侵入的方式解決微服務場景下面臨的分布式事務問題，它目前提供AT模式(即2PC)及TCC模式的分布式事務解決方案，

這里不重點介紹

可以參考seata實作2PC事務：https://blog.csdn.net/weixin_41979002/article/details/120198259

1.2.4 TCC方案

TCC是Try、Confirm、Cancel三個詞語的縮寫，TCC要求每個分支事務實作三個操作：預處理Try、確認 Confirm、撤銷Cancel，Try操作做業務檢查及資源預留，Confirm做業務確認操作，Cancel實作一個與Try相反的操作即回滾操作，TM首先發起所有的分支事務的try操作，任何一個分支事務的try操作執行失敗，TM將會發起所有分支事務的Cancel操作，若try操作全部成功，TM將會發起所有分支事務的Confirm操作，其中Confirm/Cancel 操作若執行失敗，TM會進行重試，

這里不重點介紹

1.2.5 本地訊息表方案

可靠訊息最終一致性方案是指當事務發起方執行完成本地事務后并發出一條訊息，事務參與方(訊息消費者)一定能夠接收訊息并處理事務成功，此方案強調的是只要訊息發給事務參與方最終事務要達到一致，

此方式通過mq中間件實作，也是本文重點介紹方式

1.2.6 RocketMQ事務訊息方案

這里不做過多介紹，這里講的是rabbitmq；但兩者實作方式稍有略同，

1.2.7 最大努力通知

你們百度吧，

1.3 小結

上述介紹了多個解決方案，沒有最好的只有最合適的，在不同的業務場景，可以選擇不同的實作方案，視情況而定，

二、rabbitmq本地訊息表方案

2.1 方案流程介紹

本地訊息表這個方案最初是eBay提出的，此方案的核心是通過本地事務保證資料業務操作和訊息的一致性，然后通過定時任務將訊息發送至訊息中間件，待確認訊息發送給消費方成功再將訊息洗掉，
在這里插入圖片描述

互動流程如下：
1、用戶注冊
用戶服務在本地事務新增用戶和增加 ”積分訊息日志“，（用戶表和訊息表通過本地事務保證一致）下邊是偽代碼

begin transaction； 
	//1.新增用戶 
	//2.存盤積分訊息日志 
commit transation;

這種情況下，本地資料庫操作與存盤積分訊息日志處于同一個事務中，本地資料庫操作與記錄訊息日志操作具備原子性，
2、定時任務掃描日志
如何保證將訊息發送給訊息佇列呢？

經過第一步訊息已經寫到訊息日志表中，可以啟動獨立的執行緒，定時對訊息日志表中的訊息進行掃描并發送至訊息中間件，在訊息中間件反饋發送成功后洗掉該訊息日志，否則等待定時任務下一周期重試，

3、消費訊息

如何保證消費者一定能消費到訊息呢？

這里可以使用MQ的ack（即訊息確認）機制，消費者監聽MQ，如果消費者接收到訊息并且業務處理完成后向MQ 發送ack（即訊息確認），此時說明消費者正常消費訊息完成，MQ將不再向消費者推送訊息，否則消費者會不斷重試向消費者來發送訊息，

積分服務接收到”增加積分“訊息，開始增加積分，積分增加成功后向訊息中間件回應ack，否則訊息中間件將重復投遞此訊息，

由于訊息會重復投遞，積分服務的”增加積分“功能需要實作冪等性，

2.2 生產者端的發送可靠保證

2.2.0 創建生產者資料庫表

在這里插入圖片描述
一個是主要業務資料表，一個是事務日志記錄表，

2.2.1 創建專案工程

在這里插入圖片描述

2.2.2 添加依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>dtx-demo</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>dtx-txmsg-demo-bank1</artifactId>
    <dependencies>
        <!--rabbitmq-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

        <!--springboot+springcloud基本依賴-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-openfeign</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-hystrix</artifactId>
        </dependency>

        <!--日志列印selfj-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>

        <!--mybatis+mysql+資料源連接依賴-->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid-spring-boot-starter</artifactId>
            <version>1.1.10</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
        </dependency>

        <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.62</version>
        </dependency>
    </dependencies>
</project>

2.2.3 application.yml

server:
  port: 8001
spring:
  datasource:
    #當前資料源操作型別
    type: com.alibaba.druid.pool.DruidDataSource
    #mysql驅動包
    driver-class-name: org.gjt.mm.mysql.Driver
    url: jdbc:mysql://localhost:3306/bank1?useUnicode=true&characterEncoding=utf8&useSSL=false
    username: root
    password: 123456

  rabbitmq:
    username: guest
    password: guest
    virtual-host: /
    addresses: 192.168.229.128:5672
    publisher-confirm-type: correlated
    publisher-returns: true

2.2.4 創建交換機和佇列

package cn.itcast.dtx.txmsgdemo.bank1.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Configuration
public class DirectRabbitConfig {

    //創建佇列
    @Bean
    public Queue directEmailQueue() {
        // durable:是否持久化,默認是false,持久化佇列：會被存盤在磁盤上，當訊息代理重啟時仍然存在，暫存佇列：當前連接有效
        // exclusive:默認也是false，只能被當前創建的連接使用，而且當連接關閉后佇列即被洗掉，此參考優先級高于durable
        // autoDelete:是否自動洗掉，當沒有生產者或者消費者使用此佇列，該佇列會自動洗掉，
        //一般設定一下佇列的持久化就好,其余兩個就是默認false
        return new Queue("account.queue", true);
    }

    //創建交換機
    @Bean
    public DirectExchange directOrderExchange() {
        return new DirectExchange("account_exchange", true, false);
    }

    //系結關系
    @Bean
    public Binding directEmailBinding() {
        return BindingBuilder.bind(directEmailQueue()).to(directOrderExchange()).with("account");
    }
}

2.2.5 dao

package cn.itcast.dtx.txmsgdemo.bank1.dao;

import cn.itcast.dtx.txmsgdemo.bank1.entity.AccountInfo;
import cn.itcast.dtx.txmsgdemo.bank1.entity.AccountInfoLog;
import cn.itcast.dtx.txmsgdemo.bank1.model.AccountChangeEvent;
import cn.itcast.dtx.txmsgdemo.bank1.schedule.ScanAccountLog;
import org.apache.ibatis.annotations.*;
import org.springframework.stereotype.Component;

import java.util.List;

@Mapper
@Component
public interface AccountInfoDao {
    @Update("update account_info set account_balance=account_balance-#{amount}  where account_no=#{accountNo}")
    int updateAccountBalance(@Param("accountNo") String accountNo, @Param("amount") Double amount);

    @Select("select count(1) from de_duplication where tx_no = #{txNo}")
    int isExistTx(String txNo);


    @Insert("insert into de_duplication values(#{txNo},now());")
    int addTx(String txNo);

    @Insert("insert into account_info_log values(null,#{accountNo},#{amount},#{txNo},'Y','2');")
    int addAccountLog(@Param("accountNo") String accountNo, @Param("amount") Double amount,@Param("txNo") String txNo);

    @Update("update account_info_log set status = #{status} where txNo = #{txNo};")
    int updAccountLog(@Param("txNo") String txNo,@Param("status") String status);

    @Select("select * from account_info_log where status = 'N'")
    List<AccountInfoLog> selectFailAcount();
}

2.2.6 訊息發送確認方法

package cn.itcast.dtx.txmsgdemo.bank1.config;

import cn.itcast.dtx.txmsgdemo.bank1.dao.AccountInfoDao;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class ConfirmService implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {

    @Autowired
    AccountInfoDao accountInfoDao;
    
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("訊息標識：" + correlationData.toString());
        log.info("發送成功確認："+ack);
        log.info("錯誤原因："+cause);

        //該方法在訊息到達MQ服務器時都表示成功，并不能保證訊息一定會被投遞到目標 queue 里，所以這個方法不適合做確認
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("訊息："+message.toString());
        log.info("回傳碼："+replyCode);
        log.info("回傳描述："+replyText);
        log.info("交換機："+exchange);
        log.info("路由key："+routingKey);

        //方法執行，說明訊息未投遞成功，將日志表的狀態修改為N
        String msg = new String(message.getBody());
        accountInfoDao.updAccountLog((String) JSONObject.parseObject(msg).get("txNo"),"N");
    }
}

2.2.7 核心業務方法

package cn.itcast.dtx.txmsgdemo.bank1.service.impl;

import cn.itcast.dtx.txmsgdemo.bank1.config.ConfirmService;
import cn.itcast.dtx.txmsgdemo.bank1.dao.AccountInfoDao;
import cn.itcast.dtx.txmsgdemo.bank1.model.AccountChangeEvent;
import cn.itcast.dtx.txmsgdemo.bank1.service.AccountInfoService;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.UUID;

/**
 * @author Administrator
 * @version 1.0
 **/
@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {

    @Autowired
    AccountInfoDao accountInfoDao;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private ConfirmService confirmService;

    @Override
    @Transactional
    public void doUpdateAccountBalance(AccountChangeEvent accountChangeEvent) throws Exception {

        //本地扣減金額
        int i = accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(), accountChangeEvent.getAmount());
        if (i != 1){
            throw new Exception("更新失敗");
        }

        //記錄本地事務日志表
        int j = accountInfoDao.addAccountLog(accountChangeEvent.getAccountNo(), accountChangeEvent.getAmount(),accountChangeEvent.getTxNo());
        if (j != 1){
            throw new Exception("插入失敗");
        }

        //發送訊息到佇列
        log.info("發送訊息："+JSONObject.toJSONString(accountChangeEvent));
        rabbitTemplate.setConfirmCallback(confirmService);
        rabbitTemplate.setReturnCallback(confirmService);
        rabbitTemplate.convertAndSend("account_exchange", "account1", JSONObject.toJSONString(accountChangeEvent), message -> {
                    message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                    return message;
                },new CorrelationData(accountChangeEvent.getTxNo()));
    }
}

2.2.8 定時任務

package cn.itcast.dtx.txmsgdemo.bank1.schedule;

import cn.itcast.dtx.txmsgdemo.bank1.config.ConfirmService;
import cn.itcast.dtx.txmsgdemo.bank1.dao.AccountInfoDao;
import cn.itcast.dtx.txmsgdemo.bank1.entity.AccountInfoLog;
import com.alibaba.fastjson.JSONObject;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.UUID;

@Component
public class ScanAccountLog {

    @Autowired
    private AccountInfoDao accountInfoDao;

    @Autowired
    private ConfirmService confirmService;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Scheduled(initialDelay = 1000,fixedDelay = 5000)
    @Transactional
    public void scan(){
        List<AccountInfoLog> accountInfoLogs = accountInfoDao.selectFailAcount();
        if (!CollectionUtils.isEmpty(accountInfoLogs)){

            accountInfoLogs.forEach(accountInfoLog -> {
                System.out.println("定時任務執行");
                accountInfoDao.updAccountLog(accountInfoLog.getTxNo(),"Y");

                rabbitTemplate.setConfirmCallback(confirmService);
                rabbitTemplate.setReturnCallback(confirmService);
                rabbitTemplate.convertAndSend("account_exchange", "account", JSONObject.toJSONString(accountInfoLog), message -> {
                            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                            return message;
                        },new CorrelationData(accountInfoLog.getTxNo()));
            });
        }

    }
}

2.2.9 controller

package cn.itcast.dtx.txmsgdemo.bank1.controller;

import cn.itcast.dtx.txmsgdemo.bank1.model.AccountChangeEvent;
import cn.itcast.dtx.txmsgdemo.bank1.service.AccountInfoService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;

/**
 * @author Administrator
 * @version 1.0
 **/
@RestController
@Slf4j
public class AccountInfoController {
    @Autowired
    private AccountInfoService accountInfoService;

    @GetMapping(value = "/transfer")
    public String transfer( @RequestParam("amount") Double amount) throws Exception {
        //創建一個事務id，作為訊息內容發到mq
        String tx_no = UUID.randomUUID().toString();
        AccountChangeEvent accountChangeEvent = new AccountChangeEvent("1",amount,tx_no,"2");
        accountInfoService.doUpdateAccountBalance(accountChangeEvent);
        return "轉賬成功";
    }
}

2.2.10 主啟動

package cn.itcast.dtx.txmsgdemo.bank1;


import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class TxMsgBank1Service {

	public static void main(String[] args) {
		SpringApplication.run(TxMsgBank1Service.class, args);
	}

}

2.2.11 小結

生產者端主要保證訊息能發到佇列里，不需要考慮回滾，只要保證訊息發到佇列里，其他的就不用管了，（因此需要視場景而定）

需要注意的幾個點：
1、日志表的作用主要是用于記錄事務記錄，備份請求的操作相關資料

2、業務方法：執行轉賬方法時，更新本地賬戶資料表，并插入事務日志表，初始狀態設定為Y代表發送成功，若出錯時會執行訊息回退方法returnedMessage，（簡介：returnedMessage在交換機發送訊息到佇列時出錯執行，confirm在訊息從生產者到mq服務即會執行無論訊息是否能成功發送到佇列，若前者出錯returnedMessage總會在confirm之前執行，因此這里只需要關注returnedMessage方法即可）

3、訊息確認方法作用是：若發送訊息出錯時會執行，將本地日志表記錄更改為未發送成功狀態

4、定時任務：用于掃描事務記錄表，掃描未成功狀態的訊息資料，重新呼叫發送訊息方法，直到訊息發送成功，提供訊息發送可靠服務，

2.3 消費者端消費可靠保證

2.3.0 創建消費者資料庫

在這里插入圖片描述
一個是主要業務資料表，一個是事務冪等控制表，

2.3.1 創建生產者工程

在這里插入圖片描述
有些類是多余的，參考你需要的類即可，

2.3.2 依賴

參考生產者

2.3.3 yml

server:
  port: 8002

spring:
  datasource:
    #當前資料源操作型別
    type: com.alibaba.druid.pool.DruidDataSource
    #mysql驅動包
    driver-class-name: org.gjt.mm.mysql.Driver
    url: jdbc:mysql://localhost:3306/bank2?useUnicode=true&characterEncoding=utf8&useSSL=false
    username: root
    password: 123456
  rabbitmq:
    username: guest
    password: guest
    virtual-host: /
    addresses: 192.168.229.128:5672
    listener:
      simple:
        acknowledge-mode: manual # 設定消費端手動 ack
        retry:
          enabled: true # 是否支持重試

2.3.4 dao

package cn.itcast.dtx.txmsgdemo.bank2.dao;

import org.apache.ibatis.annotations.*;
import org.springframework.stereotype.Component;

import java.util.List;

@Mapper
@Component
public interface AccountInfoDao {
    @Update("update account_info set account_balance=account_balance+#{amount}  where account_no=#{otherNo}")
    int updateAccountBalance(@Param("otherNo") String otherNo, @Param("amount") Double amount);

    @Select("select count(1) from de_duplication where tx_no = #{txNo}")
    int isExistTx(String txNo);

    @Insert("insert into de_duplication values(#{txNo},now());")
    int addTx(String txNo);
}

2.3.5 消費者方法

package cn.itcast.dtx.txmsgdemo.bank2.config;

import cn.itcast.dtx.txmsgdemo.bank2.dao.AccountInfoDao;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;

@RabbitListener(queues = {"account.queue"})
@Component
public class TxListener {

    @Autowired
    private AccountInfoDao accountInfoDao;

    @RabbitHandler
    @Transactional
    public void messagerevice(String msg, Channel channel, Message message) throws IOException {
        try {
            JSONObject jsonObject = JSONObject.parseObject(msg);

            //冪等檢驗
            if (accountInfoDao.isExistTx(jsonObject.getString("txNo"))>0){
                throw new Exception("事務id為"+jsonObject.getString("txNo")+"已經消費過了");
            }
            accountInfoDao.addTx(jsonObject.getString("txNo"));

            //更新賬戶
            int i = accountInfoDao.updateAccountBalance(jsonObject.getString("otherNo"), jsonObject.getDouble("amount"));
            if (i != 1){
                throw new Exception("更新失敗");
            }

            //消費訊息確認應答
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }catch (Exception e){
            //重試消費，或投遞到死信佇列
            //注意：引數三是否運行重試，若設定為true，會出現死回圈，你可以定義常量設定重試次數
            channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);

            e.printStackTrace();
        }
    }
}

2.3 6 主啟動

package cn.itcast.dtx.txmsgdemo.bank2;


import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class TxMsgBank2Service {

	public static void main(String[] args) {
		SpringApplication.run(TxMsgBank2Service.class, args);
	}

}

2.3.7 小結

消費者端實作起來比較簡單，

需要要主要的幾點：

1、消費者消費時需要考慮冪等問題，因為可能會出現重復消費的問題
2、若消費者執行程序出錯時，進入catch代碼塊，若執行了basicNack方法，mq會收到應答后訊息會被移除，這時你可以視業務場景，嘗試重復消費或者進入死信佇列，
3、設定了手動應答后，在yml配置重試次數將不會生效，只能通過代碼實作，
4、訊息消費成功，可以發送一個訊息給生產者，生產者再去清空事務記錄表，（選做）

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/333620.html

標籤：其他

上一篇：數倉搭建 ODS 層

下一篇：訊息中間件MQ及ActiveMQ介紹

Rabbitmq的分布式事務解決方案（十二）

一、前言

1.1 簡述

1.2 分布式事務的幾種解決方案

1.2.1 2PC(兩階段提交)

1.2.2 XA方案

1.2.3 Seata方案

1.2.4 TCC方案

1.2.5 本地訊息表方案

1.2.6 RocketMQ事務訊息方案

1.2.7 最大努力通知

1.3 小結

二、rabbitmq本地訊息表方案

2.1 方案流程介紹

2.2 生產者端的發送可靠保證

2.2.0 創建生產者資料庫表

2.2.1 創建專案工程

2.2.2 添加依賴

2.2.3 application.yml

2.2.4 創建交換機和佇列

2.2.5 dao

2.2.6 訊息發送確認方法

2.2.7 核心業務方法

2.2.8 定時任務

2.2.9 controller

2.2.10 主啟動

2.2.11 小結

2.3 消費者端消費可靠保證

2.3.0 創建消費者資料庫

2.3.1 創建生產者工程

2.3.2 依賴

2.3.3 yml

2.3.4 dao

2.3.5 消費者方法

2.3 6 主啟動

2.3.7 小結