分布式事務(wù)概念

討論分布式事務(wù)之前我們分清兩個(gè)概念：本地事務(wù)、分布式事務(wù)；

本地事務(wù)是解決單個(gè)數(shù)據(jù)源上的數(shù)據(jù)操作的一致性問(wèn)題的話，而分布式事務(wù)則是為了解決跨越多個(gè)數(shù)據(jù)源上數(shù)據(jù)操作的一致性問(wèn)題。

百度官方對(duì)分布式事務(wù)的定義是指事務(wù)的參與者、支持事務(wù)的服務(wù)器、資源服務(wù)器以及事務(wù)管理器分別位于不同的分布式系統(tǒng)的不同節(jié)點(diǎn)之上。

也就是說(shuō)我們?cè)诓僮饕粋€(gè)業(yè)務(wù)邏輯過(guò)程中，涉及兩個(gè)數(shù)據(jù)源（A、B），且很多時(shí)候A、B這兩個(gè)數(shù)據(jù)源屬于兩個(gè)不同的物理環(huán)境。當(dāng)我們操作A數(shù)據(jù)源過(guò)程中出現(xiàn)異常情況，那么必須讓針對(duì)B數(shù)據(jù)源的操作回滾，同時(shí)A數(shù)據(jù)源的操作也回滾。

在Java開(kāi)發(fā)過(guò)程中事務(wù)一般使用Spring為我們提供了方便的聲明式事務(wù)方法@transactional。但是默認(rèn)的Spring事務(wù)只支持單數(shù)據(jù)源，而實(shí)際上一個(gè)系統(tǒng)往往需要寫(xiě)多個(gè)數(shù)據(jù)源，這個(gè)時(shí)候我們就需要考慮如何通過(guò)Spring實(shí)現(xiàn)對(duì)分布式事務(wù)的支持。
SpringBoot官方提供推薦了Atomikos和 Bitronix兩種無(wú)需服務(wù)器支持的分布式事務(wù)組件

JAVA領(lǐng)域中針對(duì)分布式事務(wù)的解決方案就是JTA(即Java Transaction API)；

XA和JTA概述

XA 是由 X/Open 組織提出的分布式事務(wù)的一種協(xié)議（或者稱之為分布式架構(gòu)）。它主要定義了兩部分的管理器，全局事務(wù)管理器及資源管理器。在 XA 的設(shè)計(jì)理念中，把不同資源納入到一個(gè)事務(wù)管理器進(jìn)行統(tǒng)一管理，例如數(shù)據(jù)庫(kù)資源，消息中間件資源等，從而進(jìn)行全部資源的事務(wù)提交或者取消，目前主流的數(shù)據(jù)庫(kù)，消息中間件都支持 XA 協(xié)議。

JTA 叫做 Java Transaction API，它是 XA 協(xié)議的 JAVA 實(shí)現(xiàn)。目前在 JAVA 里面，關(guān)于 JTA 的定義主要是兩部分

• 事務(wù)管理器接口-----javax.transaction.TransactionManager
• 資源管理器接口-----javax.transaction.xa.XAResource

在一般應(yīng)用采用 JTA 接口實(shí)現(xiàn)事務(wù)，需要一個(gè)外置的 JTA 容器來(lái)存儲(chǔ)這些事務(wù)，像 Tomcat。今天我們要講的是 Atomikos，它是一個(gè)獨(dú)立實(shí)現(xiàn)了 JTA 的框架，能夠在我們的應(yīng)用服務(wù)器中運(yùn)行 JTA 事務(wù)。

SpringBoot集成atomikos

數(shù)據(jù)庫(kù)結(jié)構(gòu)

CREATE TABLE `tb_order` (
  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '訂單id',
  `user_id` bigint DEFAULT NULL COMMENT '用戶id',
  `name` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci DEFAULT NULL COMMENT '商品名稱',
  `price` bigint DEFAULT NULL COMMENT '商品價(jià)格',
  `num` int DEFAULT '0' COMMENT '商品數(shù)量',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=137 DEFAULT CHARSET=utf8mb3 ROW_FORMAT=COMPACT;

添加了username為唯一索引，方便后面測(cè)試多數(shù)據(jù)插庫(kù)異常事務(wù)回滾

CREATE TABLE `tb_user` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `username` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci DEFAULT NULL COMMENT '收件人',
  `address` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci DEFAULT NULL COMMENT '地址',
  PRIMARY KEY (`id`),
  UNIQUE KEY `idx_uername` (`username`)
) ENGINE=InnoDB AUTO_INCREMENT=34 DEFAULT CHARSET=utf8mb3 ROW_FORMAT=COMPACT;

pom

<parent>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-parent</artifactId>
	<version>2.3.2.RELEASE</version>
</parent>

<dependencies>
    <!-- druid-->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>druid</artifactId>
        <version>1.1.10</version>
    </dependency>
    <!-- druid-->
    <!-- mysql-connector-->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
       <version>8.0.33</version>
    </dependency>
    <!-- mysql-connector-->
    <!-- mybatis-plus-->
    <dependency>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-boot-starter</artifactId>
        <version>3.1.1</version>
    </dependency>
    <!-- mybatis-plus-->

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jta-atomikos</artifactId>
    </dependency>

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>com.alibaba.fastjson2</groupId>
        <artifactId>fastjson2</artifactId>
        <version>2.0.25</version>
    </dependency>
    <dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>32.0.0-jre</version>
    </dependency>
</dependencies>


<build>
    <resources>
        <resource>
            <directory>src/main/java</directory>
            <includes>
                <include>**/*.xml</include>
            </includes>
            <filtering>false</filtering>
        </resource>
    </resources>
</build>

通用工具類

R

/**
 * @ClassName: R
 * @Description: 統(tǒng)一返回實(shí)體
 */
@Getter
@Setter
@SuppressWarnings({"AlibabaClassNamingShouldBeCamel"})
@Accessors(chain = true)
public class R<T> {
    public static final String DEF_ERROR_MESSAGE = "系統(tǒng)繁忙，請(qǐng)稍候再試";
    public static final String HYSTRIX_ERROR_MESSAGE = "請(qǐng)求超時(shí)，請(qǐng)稍候再試";
    public static final int SUCCESS_CODE = 0;
    public static final int FAIL_CODE = -1;
    public static final int TIMEOUT_CODE = -2;
    /**
     * 統(tǒng)一參數(shù)驗(yàn)證異常
     */
    public static final int VALID_EX_CODE = -9;
    public static final int OPERATION_EX_CODE = -10;
    /**
     * 調(diào)用是否成功標(biāo)識(shí)，0：成功，-1:系統(tǒng)繁忙，此時(shí)請(qǐng)開(kāi)發(fā)者稍候再試 詳情見(jiàn)[ExceptionCode]
     */
    private int code;

    /**
     * 調(diào)用結(jié)果
     */
    private T data;

    /**
     * 結(jié)果消息，如果調(diào)用成功，消息通常為空T
     */
    private String msg = "ok";


    private String path;
    /**
     * 附加數(shù)據(jù)
     */
    private Map<String, Object> extra;

    /**
     * 響應(yīng)時(shí)間
     */
    private long timestamp = System.currentTimeMillis();

    private R() {
        super();
    }

    public R(int code, T data, String msg) {
        this.code = code;
        this.data = data;
        this.msg = msg;
    }

    public static <E> R<E> result(int code, E data, String msg) {
        return new R<>(code, data, msg);
    }

    /**
     * 請(qǐng)求成功消息
     *
     * @param data 結(jié)果
     * @return RPC調(diào)用結(jié)果
     */
    public static <E> R<E> success(E data) {
        return new R<>(SUCCESS_CODE, data, "ok");
    }

    public static R<Boolean> success() {
        return new R<>(SUCCESS_CODE, true, "ok");
    }

    /**
     * 請(qǐng)求成功方法 ，data返回值，msg提示信息
     *
     * @param data 結(jié)果
     * @param msg  消息
     * @return RPC調(diào)用結(jié)果
     */
    public static <E> R<E> success(E data, String msg) {
        return new R<>(SUCCESS_CODE, data, msg);
    }

    /**
     * 請(qǐng)求失敗消息
     *
     * @param msg
     * @return
     */
    public static <E> R<E> fail(int code, String msg) {
        return new R<>(code, null, (msg == null || msg.isEmpty()) ? DEF_ERROR_MESSAGE : msg);
    }

    public static <E> R<E> fail(String msg) {
        return fail(OPERATION_EX_CODE, msg);
    }

    public static <E> R<E> fail(String msg, Object... args) {
        String message = (msg == null || msg.isEmpty()) ? DEF_ERROR_MESSAGE : msg;
        return new R<>(OPERATION_EX_CODE, null, String.format(message, args));
    }

    public static <E> R<E> fail(BaseExceptionCode exceptionCode) {
        return validFail(exceptionCode);
    }

    public static <E> R<E> fail(BizException exception) {
        if (exception == null) {
            return fail(DEF_ERROR_MESSAGE);
        }
        return new R<>(exception.getCode(), null, exception.getMessage());
    }

    /**
     * 請(qǐng)求失敗消息，根據(jù)異常類型，獲取不同的提供消息
     *
     * @param throwable 異常
     * @return RPC調(diào)用結(jié)果
     */
    public static <E> R<E> fail(Throwable throwable) {
        return fail(FAIL_CODE, throwable != null ? throwable.getMessage() : DEF_ERROR_MESSAGE);
    }

    public static <E> R<E> validFail(String msg) {
        return new R<>(VALID_EX_CODE, null, (msg == null || msg.isEmpty()) ? DEF_ERROR_MESSAGE : msg);
    }

    public static <E> R<E> validFail(String msg, Object... args) {
        String message = (msg == null || msg.isEmpty()) ? DEF_ERROR_MESSAGE : msg;
        return new R<>(VALID_EX_CODE, null, String.format(message, args));
    }

    public static <E> R<E> validFail(BaseExceptionCode exceptionCode) {
        return new R<>(exceptionCode.getCode(), null,
                (exceptionCode.getMsg() == null || exceptionCode.getMsg().isEmpty()) ? DEF_ERROR_MESSAGE : exceptionCode.getMsg());
    }

    public static <E> R<E> timeout() {
        return fail(TIMEOUT_CODE, HYSTRIX_ERROR_MESSAGE);
    }


    public R<T> put(String key, Object value) {
        if (this.extra == null) {
            this.extra = Maps.newHashMap();
        }
        this.extra.put(key, value);
        return this;
    }

    /**
     * 邏輯處理是否成功
     *
     * @return 是否成功
     */
    public Boolean getIsSuccess() {
        return this.code == SUCCESS_CODE || this.code == 200;
    }

    /**
     * 邏輯處理是否失敗
     *
     * @return
     */
    public Boolean getIsError() {
        return !getIsSuccess();
    }

    @Override
    public String toString() {
        return JSONObject.toJSONString(this);
    }
}

BaseController

/**
 * @ClassName: BaseController
 * @Description: controller 抽象基類
 */
public abstract class BaseController {
    /**
     * 成功返回
     *
     * @param data
     * @return
     */
    public <T> R<T> success(T data) {
        return R.success(data);
    }

    public R<Boolean> success() {
        return R.success();
    }

    /**
     * 失敗返回
     *
     * @param msg
     * @return
     */
    public <T> R<T> fail(String msg) {
        return R.fail(msg);
    }

    public <T> R<T> fail(String msg, Object... args) {
        return R.fail(msg, args);
    }

    /**
     * 失敗返回
     *
     * @param code
     * @param msg
     * @return
     */
    public <T> R<T> fail(int code, String msg) {
        return R.fail(code, msg);
    }

    public <T> R<T> fail(BaseExceptionCode exceptionCode) {
        return R.fail(exceptionCode);
    }

    public <T> R<T> fail(BizException exception) {
        return R.fail(exception);
    }

    public <T> R<T> fail(Throwable throwable) {
        return R.fail(throwable);
    }

    public <T> R<T> validFail(String msg) {
        return R.validFail(msg);
    }

    public <T> R<T> validFail(String msg, Object... args) {
        return R.validFail(msg, args);
    }

    public <T> R<T> validFail(BaseExceptionCode exceptionCode) {
        return R.validFail(exceptionCode);
    }
}

BaseExceptionCode

/**
 * @ClassName: BaseExceptionCode
 * @Description: 公共異常編碼類
 */
public interface BaseExceptionCode {
    /**
     * 異常編碼
     *
     * @return
     */
    int getCode();

    /**
     * 異常消息
     * @return
     */
    String getMsg();
}

ExceptionCode

/**
 * 全局錯(cuò)誤碼 10000-15000
 * <p>
 * 預(yù)警異常編碼    范圍： 30000~34999
 * 標(biāo)準(zhǔn)服務(wù)異常編碼 范圍：35000~39999
 * 郵件服務(wù)異常編碼 范圍：40000~44999
 * 短信服務(wù)異常編碼 范圍：45000~49999
 * 權(quán)限服務(wù)異常編碼 范圍：50000-59999
 * 文件服務(wù)異常編碼 范圍：60000~64999
 * 日志服務(wù)異常編碼 范圍：65000~69999
 * 消息服務(wù)異常編碼 范圍：70000~74999
 * 開(kāi)發(fā)者平臺(tái)異常編碼 范圍：75000~79999
 * 搜索服務(wù)異常編碼 范圍：80000-84999
 * 共享交換異常編碼 范圍：85000-89999
 * 移動(dòng)終端平臺(tái) 異常碼 范圍：90000-94999
 * <p>
 * 安全保障平臺(tái)    范圍：        95000-99999
 * 軟硬件平臺(tái) 異常編碼 范圍：    100000-104999
 * 運(yùn)維服務(wù)平臺(tái) 異常編碼 范圍：  105000-109999
 * 統(tǒng)一監(jiān)管平臺(tái)異常 編碼 范圍：  110000-114999
 * 認(rèn)證方面的異常編碼  范圍：115000-115999
 *
 */
public enum ExceptionCode implements BaseExceptionCode {

    //系統(tǒng)相關(guān) start
    SUCCESS(0, "成功"),
    SYSTEM_BUSY(-1, "系統(tǒng)繁忙~請(qǐng)稍后再試~"),
    SYSTEM_TIMEOUT(-2, "系統(tǒng)維護(hù)中~請(qǐng)稍后再試~"),
    PARAM_EX(-3, "參數(shù)類型解析異常"),
    SQL_EX(-4, "運(yùn)行SQL出現(xiàn)異常"),
    NULL_POINT_EX(-5, "空指針異常"),
    ILLEGALA_ARGUMENT_EX(-6, "無(wú)效參數(shù)異常"),
    MEDIA_TYPE_EX(-7, "請(qǐng)求類型異常"),
    LOAD_RESOURCES_ERROR(-8, "加載資源出錯(cuò)"),
    BASE_VALID_PARAM(-9, "統(tǒng)一驗(yàn)證參數(shù)異常"),
    OPERATION_EX(-10, "操作異常"),


    OK(200, "OK"),
    BAD_REQUEST(400, "錯(cuò)誤的請(qǐng)求"),
    /**
     * {@code 401 Unauthorized}.
     *
     * @see <a >HTTP/1.1: Authentication, section 3.1</a>
     */
    UNAUTHORIZED(401, "未經(jīng)授權(quán)"),
    /**
     * {@code 404 Not Found}.
     *
     * @see <a >HTTP/1.1: Semantics and Content, section 6.5.4</a>
     */
    NOT_FOUND(404, "沒(méi)有找到資源"),
    METHOD_NOT_ALLOWED(405, "不支持當(dāng)前請(qǐng)求類型"),

    TOO_MANY_REQUESTS(429, "請(qǐng)求超過(guò)次數(shù)限制"),
    INTERNAL_SERVER_ERROR(500, "內(nèi)部服務(wù)錯(cuò)誤"),
    BAD_GATEWAY(502, "網(wǎng)關(guān)錯(cuò)誤"),
    GATEWAY_TIMEOUT(504, "網(wǎng)關(guān)超時(shí)"),
    //系統(tǒng)相關(guān) end

    REQUIRED_FILE_PARAM_EX(1001, "請(qǐng)求中必須至少包含一個(gè)有效文件"),
    //jwt token 相關(guān) start

    JWT_TOKEN_EXPIRED(40001, "會(huì)話超時(shí)，請(qǐng)重新登錄"),
    JWT_SIGNATURE(40002, "不合法的token，請(qǐng)認(rèn)真比對(duì) token 的簽名"),
    JWT_ILLEGAL_ARGUMENT(40003, "缺少token參數(shù)"),
    JWT_GEN_TOKEN_FAIL(40004, "生成token失敗"),
    JWT_PARSER_TOKEN_FAIL(40005, "解析token失敗"),
    JWT_USER_INVALID(40006, "用戶名或密碼錯(cuò)誤"),
    JWT_USER_ENABLED(40007, "用戶已經(jīng)被禁用！"),
    //jwt token 相關(guān) end

    ;

    private int code;
    private String msg;

    ExceptionCode(int code, String msg) {
        this.code = code;
        this.msg = msg;
    }

    @Override
    public int getCode() {
        return code;
    }

    @Override
    public String getMsg() {
        return msg;
    }


    public ExceptionCode build(String msg, Object... param) {
        this.msg = String.format(msg, param);
        return this;
    }

    public ExceptionCode param(Object... param) {
        msg = String.format(msg, param);
        return this;
    }
}

BaseException

/**
 * @ClassName: BaseException
 * @Description: 異常接口類
 */
public interface BaseException {

    /**
     * 統(tǒng)一參數(shù)驗(yàn)證異常碼
     */
    int BASE_VALID_PARAM = -9;

    /**
     * 返回異常信息
     *
     * @return
     */
    String getMessage();

    /**
     * 返回異常編碼
     *
     * @return
     */
    int getCode();
}

BaseUncheckedException

/**
 * @ClassName: BaseUncheckedException
 * @Description: 非運(yùn)行期異?；?，所有自定義非運(yùn)行時(shí)異常繼承該類
 */
public class BaseUncheckedException extends RuntimeException implements BaseException {

    private static final long serialVersionUID = -778887391066124051L;

    /**
     * 異常信息
     */
    protected String message;

    /**
     * 具體異常碼
     */
    protected int code;

    public BaseUncheckedException(int code, String message) {
        super(message);
        this.code = code;
        this.message = message;
    }

    public BaseUncheckedException(int code, String format, Object... args) {
        super(String.format(format, args));
        this.code = code;
        this.message = String.format(format, args);
    }


    @Override
    public String getMessage() {
        return message;
    }
    @Override
    public int getCode() {
        return code;
    }
}

BizException

/**
 * @ClassName: BizException
 * @Description: 業(yè)務(wù)異常 用于在處理業(yè)務(wù)邏輯時(shí)，進(jìn)行拋出的異常。
 */
public class BizException extends BaseUncheckedException {

    private static final long serialVersionUID = -3843907364558373817L;

    public BizException(String message) {
        super(-1, message);
    }

    public BizException(int code, String message) {
        super(code, message);
    }

    public BizException(int code, String message, Object... args) {
        super(code, message, args);
    }

    /**
     * 實(shí)例化異常
     *
     * @param code    自定義異常編碼
     * @param message 自定義異常消息
     * @param args    已定義異常參數(shù)
     * @return
     */
    public static BizException wrap(int code, String message, Object... args) {
        return new BizException(code, message, args);
    }

    public static BizException wrap(String message, Object... args) {
        return new BizException(-1, message, args);
    }

    public static BizException validFail(String message, Object... args) {
        return new BizException(-9, message, args);
    }

    public static BizException wrap(BaseExceptionCode ex) {
        return new BizException(ex.getCode(), ex.getMsg());
    }

    @Override
    public String toString() {
        return "BizException [message=" + message + ", code=" + code + "]";
    }
}

application.yml

spring:
  datasource:
    druid:
      order:
        driver-class-name: com.mysql.cj.jdbc.Driver
        url: jdbc:mysql://localhost:3306/cloud_order?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&nullCatalogMeansCurrent=true
        user: root
        password: root
      user:
        driver-class-name: com.mysql.cj.jdbc.Driver
        url: jdbc:mysql://localhost:3306/cloud_user?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&nullCatalogMeansCurrent=true
        user: root
        password: root



mybatis-plus:
  #mybatis日志
  configuration:
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

logging:
  level:
    cn.zysheep.dao: debug

數(shù)據(jù)源配置類

OrderXADataSourceConfig

/**
 * @ClassName: OrderXADataSourceConfig
 * @Description: mybatis配置類 Order
 */
@Configuration
@MapperScan(basePackages = OrderXADataSourceConfig.BASE_PACKAGES, sqlSessionTemplateRef = "orderSqlSessionTemplate")
public class OrderXADataSourceConfig {
    /**
     * 掃描mapper接口包
     */
    static final String BASE_PACKAGES = "cn.zysheep.dao.order";
    /**
     * 掃描的mapper配置文件路徑
     */
    private static final String MAPPER_LOCATION = "classpath:/mapper/order/*Mapper.xml";

    /**
     * 將這個(gè)對(duì)象放入spring容器中（交給Spring管理）
     * @ConfigurationProperties 自動(dòng)配置屬性
     */
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.druid.order")
    public XADataSource getDataSourceOrder(){
        // 創(chuàng)建XA連接池
        return new MysqlXADataSource();
    }

    /**
     * 創(chuàng)建Atomikos數(shù)據(jù)源
     * 注解@DependsOn("druidXADataSourcePre")，在名為druidXADataSourcePre的bean實(shí)例化后加載當(dāng)前bean
     * @param xaDataSource
     * @return
     */
    @Bean
    @DependsOn("getDataSourceOrder")
    @Primary
    public DataSource dataSourceOrder(@Qualifier("getDataSourceOrder") XADataSource xaDataSource){
        //這里的AtomikosDataSourceBean使用的是spring提供的
        AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
        atomikosDataSourceBean.setXaDataSourceClassName("com.alibaba.druid.pool.xa.DruidXADataSource");
        // 必須為數(shù)據(jù)源指定唯一標(biāo)識(shí)
        atomikosDataSourceBean.setUniqueResourceName("dataSourceOrder");
        atomikosDataSourceBean.setPoolSize(5);
        atomikosDataSourceBean.setTestQuery("select 1");
        atomikosDataSourceBean.setBorrowConnectionTimeout(3);
        atomikosDataSourceBean.setXaDataSource(xaDataSource);

        return atomikosDataSourceBean;
    }

    /**
     * 創(chuàng)建 SqlSessionFactory
     * @return
     * @throws Exception
     */
    @Bean
    @Primary
    public SqlSessionFactory orderSqlSessionFactory(@Qualifier("dataSourceOrder") DataSource dataSource) throws Exception{
        // 用來(lái)創(chuàng)建 SqlSessionFactory 等同于下面配置
//        <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
//            <property name="dataSource" ref="dataSource" />
//            <property name="mapperLocations" value="classpath:mybatis-mapper/*.xml"/>
//        </bean>

        // 在配置sqlSession工廠類的時(shí)候，創(chuàng)建的是MybatisSqlSessionFactoryBean，是為了能夠正常使用Mybatis-Plus組件的基本功能，比如通用的crud語(yǔ)句綁定
        MybatisSqlSessionFactoryBean sqlSessionFactoryBean = new MybatisSqlSessionFactoryBean();
        sqlSessionFactoryBean.setDataSource(dataSource);
        // 設(shè)置mybatis的xml所在位置(掃描mybatis的相關(guān)xml文件，裝配到容器中)
        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources(MAPPER_LOCATION));
        //手動(dòng)設(shè)置session工廠時(shí)，需要手動(dòng)添加分頁(yè)插件
        Interceptor[] plugins = new Interceptor[1];
        plugins[0] = new PaginationInterceptor();
        sqlSessionFactoryBean.setPlugins(plugins);

        return sqlSessionFactoryBean.getObject();
    }

    /**
     * 通過(guò) SqlSessionFactory 來(lái)創(chuàng)建 SqlSessionTemplate
     * @param sqlSessionFactory
     * @return
     */
    @Bean
    @Primary
    public SqlSessionTemplate orderSqlSessionTemplate(@Qualifier("orderSqlSessionFactory") SqlSessionFactory sqlSessionFactory){
        // SqlSessionTemplate是線程安全的，可以被多個(gè)DAO所共享使用
        return new SqlSessionTemplate(sqlSessionFactory);
    }
}

UserXADataSourceConfig

/**
 * @ClassName: UserXADataSourceConfig
 * @Description: mybatis配置類 User
 */
@Configuration
@MapperScan(basePackages = UserXADataSourceConfig.BASE_PACKAGES, sqlSessionTemplateRef = "userSqlSessionTemplate")
public class UserXADataSourceConfig {
    /**
     * 掃描mapper接口包
     */
    static final String BASE_PACKAGES = "cn.zysheep.dao.user";
    /**
     * 掃描的mapper配置文件路徑
     */
    private static final String MAPPER_LOCATION = "classpath:/mapper/user/*Mapper.xml";


    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.druid.user")
    public XADataSource getDataSourceUser(){
        // 創(chuàng)建XA連接池
        return new MysqlXADataSource();
    }

    @Bean
    @DependsOn("getDataSourceUser")
    public DataSource dataSourceUser(@Qualifier("getDataSourceUser") XADataSource xaDataSource){
        //這里的AtomikosDataSourceBean使用的是spring提供的
        AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
        atomikosDataSourceBean.setUniqueResourceName("dataSourceUser");
        atomikosDataSourceBean.setXaDataSourceClassName("com.alibaba.druid.pool.xa.DruidXADataSource");
        atomikosDataSourceBean.setTestQuery("select 1");
        atomikosDataSourceBean.setBorrowConnectionTimeout(3);
        atomikosDataSourceBean.setXaDataSource(xaDataSource);
        return atomikosDataSourceBean;
    }

    /**
     * 創(chuàng)建 SqlSessionFactory
     * @return
     * @throws Exception
     */
    @Bean
    @Primary
    public SqlSessionFactory userSqlSessionFactory(@Qualifier("dataSourceUser") DataSource dataSource) throws Exception{
        // 用來(lái)創(chuàng)建 SqlSessionFactory 等同于下面配置
//        <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
//            <property name="dataSource" ref="dataSource" />
//            <property name="mapperLocations" value="classpath:mybatis-mapper/*.xml"/>
//        </bean>

        // 在配置sqlSession工廠類的時(shí)候，創(chuàng)建的是MybatisSqlSessionFactoryBean，是為了能夠正常使用Mybatis-Plus組件的基本功能，比如通用的crud語(yǔ)句綁定
        MybatisSqlSessionFactoryBean sqlSessionFactoryBean = new MybatisSqlSessionFactoryBean();
        sqlSessionFactoryBean.setDataSource(dataSource);
        // 設(shè)置mybatis的xml所在位置(掃描mybatis的相關(guān)xml文件，裝配到容器中)
        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources(MAPPER_LOCATION));
        //手動(dòng)設(shè)置session工廠時(shí)，需要手動(dòng)添加分頁(yè)插件
        Interceptor[] plugins = new Interceptor[1];
        plugins[0] = new PaginationInterceptor();
        sqlSessionFactoryBean.setPlugins(plugins);

        return sqlSessionFactoryBean.getObject();
    }

    /**
     * 通過(guò) SqlSessionFactory 來(lái)創(chuàng)建 SqlSessionTemplate
     * @param sqlSessionFactory
     * @return
     */
    @Bean
    @Primary
    public SqlSessionTemplate userSqlSessionTemplate(@Qualifier("userSqlSessionFactory") SqlSessionFactory sqlSessionFactory){
        // SqlSessionTemplate是線程安全的，可以被多個(gè)DAO所共享使用
        return new SqlSessionTemplate(sqlSessionFactory);
    }
}

1、每個(gè)數(shù)據(jù)源對(duì)應(yīng)一個(gè)配置類
2、每個(gè)配置類的@MapperScan注解不一樣，各自對(duì)應(yīng)自己mapper接口文件夾（這就是為什么要將不同數(shù)據(jù)源的mapper接口寫(xiě)在不同文件夾的原因了）
3、在配置sqlSession工廠類的時(shí)候，創(chuàng)建的是MybatisSqlSessionFactoryBean，是為了能夠正常使用Mybatis-Plus組件的基本功能，比如通用的crud語(yǔ)句綁定。
4、配置工廠類的時(shí)候，需要指定各自mapper.xml存放的路徑（這就是為什么要將不同數(shù)據(jù)源的mapper.xml寫(xiě)在不同文件夾的原因了）
5、配置工廠類的時(shí)候，需要手動(dòng)將分頁(yè)插件加進(jìn)去。因?yàn)閿?shù)據(jù)源相關(guān)的自動(dòng)配置被我們關(guān)閉了，創(chuàng)建傳統(tǒng)PaginationInterceptor類的方法已經(jīng)不好使了

實(shí)體類

Order

@Builder
@Data
@TableName("tb_order")
public class Order {
    @TableId(type = IdType.AUTO)
    private Long id;
    @TableField("user_id")
    private Long userId;
    @TableField("name")
    private String name;
    @TableField("price")
    private Long price;
    @TableField("num")
    private Integer num;
}

User

@Builder
@Data
@TableName("tb_user")
public class User {
    @TableId(type = IdType.AUTO)
    private Long id;
    @TableField("username")
    private String username;
    @TableField("address")
    private String address;
}

Mapper

OrderMapper

public interface OrderMapper extends BaseMapper<Order> {
}

配置文件 OrderMapper.xml，Mapper類路徑、和配置路徑必須數(shù)據(jù)源配置類的路徑一致否則會(huì)報(bào)錯(cuò)

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="cn.zysheep.dao.order.OrderMapper">

</mapper>

UserMapper

public interface UserMapper extends BaseMapper<User> {
}

配置文件UserMapper.xml

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="cn.zysheep.dao.user.UserMapper">

</mapper>

Service

OrderService

public interface OrderService extends IService<Order> {
    /**
     * 保存訂單
     */
    void saveOrder() throws Exception;
}

@Service
@AllArgsConstructor
public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements OrderService {

    private final OrderMapper orderMapper;

    private final UserMapper userMapper;

    /**
     * 實(shí)現(xiàn)多數(shù)據(jù)庫(kù)操作
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveOrder() throws Exception {
        // order數(shù)據(jù)源
        Order order = Order.builder()
                .userId(1000L)
                .name("Apple 蘋(píng)果 iPhone 12 ")
                .price(699900L)
                .num(1).build();
        orderMapper.insert(order);

        // user數(shù)據(jù)源
        User user = User.builder()
                .id(1001L)
                .address("長(zhǎng)沙")
                .username("封于修").build();

        userMapper.insert(user);

        // throw new Exception("12312");
    }
}

測(cè)試分布式事務(wù)

atomikos多數(shù)據(jù)源分布式事務(wù)和Spring聲明式事務(wù)使用方法一樣，類或方法加@Transactional注解。

1、正常保存

2、order數(shù)據(jù)源保存成功，user數(shù)據(jù)源保存成功，方法其他地方拋出異常，方法事務(wù)回滾

3、order數(shù)據(jù)源保存成功，user數(shù)據(jù)源保存失敗，方法事務(wù)回滾

4、order數(shù)據(jù)源保存失敗，user數(shù)據(jù)源保存不執(zhí)行，方法事務(wù)回滾

UserService

public interface UserService extends IService<User> {
    /**
     * 保存用戶
     * @throws Exception
     */
    void saveUser() throws Exception;
}

@Service
@AllArgsConstructor
public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements UserService {

    private final OrderMapper orderMapper;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveUser() throws Exception {
        // order數(shù)據(jù)源  拋出異常，方法事務(wù)回滾
        Order order = Order.builder()
                .userId(1000L)
                .name("Apple 蘋(píng)果 iPhone 12 ")
                .price(699900L)
                .num(1).build();
        orderMapper.insert(order);


        // 2、user數(shù)據(jù)源 拋出異常，方法事務(wù)回滾
        User user = User.builder()
                .id(1001L)
                .address("長(zhǎng)沙")
                .username("封于修").build();

        saveBatch(Collections.singletonList(user));

        // 1、主方法拋出異常，方法事務(wù)回滾
        // throw new Exception("12312");
    }
}

測(cè)試分布式事務(wù)

atomikos多數(shù)據(jù)源分布式事務(wù)和Spring聲明式事務(wù)使用方法一樣，類或方法加@Transactional注解。

這里主要測(cè)試Mybaits-Plus提供的批量新增是否支持 atomikos多數(shù)據(jù)源分布式事務(wù)，測(cè)試是方法內(nèi)部其他數(shù)據(jù)源發(fā)生異常事務(wù)是可以回滾的；文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-568122.html

Controller

OrderController

@RestController
@RequestMapping("/order")
@AllArgsConstructor
public class OrderController extends BaseController {

    private final OrderService orderService;

    @PostMapping("/save")
    public R save() throws Exception {
        orderService.saveOrder();
        return success();
    }
}

UserController

@RestController
@RequestMapping("/user")
@AllArgsConstructor
public class UserController extends BaseController {
    private final UserService userService;

    @PostMapping("/batchSave")
    public R batchSave() throws Exception {
        userService.saveUser();
        return success();
    }
}

啟動(dòng)類

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

到了這里，關(guān)于【SpringBoot應(yīng)用篇】SpringBoot集成atomikos實(shí)現(xiàn)多數(shù)據(jù)源配置和分布式事務(wù)管理的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！