Spring Boot+MyBatis+Atomikos+MySQL (mit Quellcode)

In tatsächlichen Projekten versuchen wir, verteilte Transaktionen zu vermeiden. Allerdings ist es manchmal wirklich notwendig, einige Dienste aufzuteilen, was zu Problemen bei verteilten Transaktionen führt.

Gleichzeitig sind verteilte Transaktionen auch eine Marktfrage in Vorstellungsgesprächen. Sie können mit diesem Fall üben und im Vorstellungsgespräch 123 sprechen.

Hier ist eine geschäftliche Kastanie: Wenn ein Benutzer einen Coupon erhält, muss die Häufigkeit, mit der der Benutzer den Coupon erhält, abgezogen werden, und dann wird eine Aufzeichnung darüber aufgezeichnet, wie oft der Benutzer den Coupon erhält.

Vor der Aufteilung

Nach der Aufteilung

Ursprünglich können Sie hier die Nachrichtenwarteschlangenmethode verwenden und asynchron verwenden, um Benutzersammlungsdatensätze hinzuzufügen. Allerdings besteht hier die Anforderung, dass Benutzer ihre Sammlungsdatensätze unmittelbar nach Erhalt einsehen können müssen. Deshalb haben wir hier Atomikos eingeführt, um verteilte Transaktionsprobleme zu implementieren.

Verteilte Transaktionen

Verteilte Transaktionen sind Transaktionen, die sich über mehrere Computer oder Datenbanken erstrecken, und es kann zu Netzwerkverzögerungen, Ausfällen oder Inkonsistenzen zwischen diesen Computern oder Datenbanken kommen. Verteilte Transaktionen müssen die Atomizität, Konsistenz, Isolation und Dauerhaftigkeit aller Vorgänge sicherstellen, um die Richtigkeit und Integrität der Daten sicherzustellen.

Was sind die verteilten Transaktionsprotokolle?

Es gibt zwei Haupttypen verteilter Transaktionsprotokolle: 2PC (Two-Phase Commit) und 3PC (Three-Phase Commit).

2PC ist derzeit das am häufigsten verwendete verteilte Transaktionsprotokoll und sein Prozess ist in zwei Phasen unterteilt: die Vorbereitungsphase und die Übermittlungsphase. In der Vorbereitungsphase gibt der Transaktionskoordinator eine Vorbereitungsanforderung an alle Teilnehmer aus, und die Teilnehmer führen die lokale Transaktion in den Vorbereitungszustand aus und geben die Vorbereitungsergebnisse an den Transaktionskoordinator zurück. Wenn in der Festschreibungsphase alle Teilnehmer erfolgreich ausgeführt werden, sendet der Transaktionskoordinator eine Festschreibungsanforderung an alle Teilnehmer und die Teilnehmer schreiben die lokale Transaktion fest. Andernfalls sendet der Transaktionskoordinator eine Rollback-Anfrage an alle Teilnehmer und die Teilnehmer führen ein Rollback der lokalen Transaktion durch Transaktionsrolle.

3PC ist eine verbesserte Version von 2PC, die eine auf 2PC basierende Vorbereitungs- und Einreichungsphase hinzufügt. In der Vorbereitungsphase für die Übermittlung fragt der Koordinator die Teilnehmer, ob sie ihre Einwilligung erteilen können. Wenn der Teilnehmer seine Einwilligung zurückgibt, wird er direkt in der Übermittlungsphase übermittelt, andernfalls wird er in der Übermittlungsphase zurückgesetzt.

Was sind die gängigen Lösungen für verteilte Transaktionen?

Zu den Implementierungslösungen für verteilte Transaktionslösungen gehören:

• Verteilte Transaktionslösungen basierend auf Nachrichtenwarteschlangen (wie die Open-Source-Lösung von RocketMQ)
• Verteilte Transaktionslösungen basierend auf verteilten Transaktions-Frameworks (wie Seata, TCC-Transaction und andere Frameworks)
• Verteilte Transaktionslösungen basierend auf dem XA-Protokoll (wie JTA usw.)
• Verteilte Transaktionslösungen basierend auf zuverlässiger Nachrichtenkonsistenz (wie Alibabas verteilte Transaktions-Middleware GTS)
• Verteilt Transaktionslösung basierend auf dem CAP-Prinzip (z. B. Event-Sourcing-Modus in der CQRS-Architektur)

Was ist JTA?

JTA (Java Transaction API) ist die Programmierschnittstellenspezifikation von J2EE. Es handelt sich um die JAVA-Implementierung des XA-Protokolls. Es definiert hauptsächlich:

Eine Transaktionsmanagerschnittstellejavax.transaction.TransactionManager，定义了有关事务的开始、提交、撤回等>操作。

一个满足XA规范的资源定义接口javax.transaction.xa.XAResource，一种资源如果要支持JTA事务，就需要让它的资源实现该XAResource

Eine Ressourcendefinitionsschnittstelle, die den XA-Spezifikationen entsprichtXAResource-Schnittstelle und implementieren Sie die von dieser Schnittstelle definierte zweiphasige Übermittlungsschnittstelle. Wenn wir eine Anwendung haben, die die JTA-Schnittstelle zum Implementieren von Transaktionen verwendet, benötigt sie beim Ausführen einen Container, der JTA implementiert. Im Allgemeinen ist dies ein J2EE-Container, wie z. B. JBoss, Websphere und andere Anwendungsserver.

Es gibt jedoch auch einige unabhängige Frameworks, die JTA implementieren. Beispielsweise stellen Atomikos und Bitronix alle JTA-Implementierungsframeworks in Form von JAR-Paketen bereit. Auf diese Weise können wir Anwendungssysteme ausführen, die JTA verwenden, um Transaktionen auf Servern wie Tomcat oder Jetty zu implementieren. 🎜

Wie oben im Unterschied zwischen lokalen Transaktionen und externen Transaktionen erwähnt, sind JTA-Transaktionen externe Transaktionen und können zur Implementierung von Transaktionalität für mehrere Ressourcen verwendet werden. Genau das macht es mit jeder Ressource XAResource来进行两阶段提交的控制。感兴趣的同学可以看看这个接口的方法，除了commit, rollback等方法以外，还有end(), forget(), isSameRM(), prepare() und noch mehr. Allein anhand dieser Schnittstellen können Sie sich die Komplexität von JTA bei der Implementierung zweiphasiger Transaktionen vorstellen.

Was ist XA?

XA ist eine verteilte Transaktionsarchitektur (oder ein Protokoll), die von der X/Open-Organisation vorgeschlagen wird. Die XA-Architektur definiert hauptsächlich die Schnittstelle zwischen dem (globalen) Transaktionsmanager und dem (lokalen) Ressourcenmanager. Die XA-Schnittstelle ist eine bidirektionale Systemschnittstelle, die eine Kommunikationsbrücke zwischen dem Transaktionsmanager und einem oder mehreren Ressourcenmanagern bildet. Mit anderen Worten: Bei einer auf XA basierenden Transaktion können wir die Transaktionsverwaltung für mehrere Ressourcen durchführen. Beispielsweise greift ein System auf mehrere Datenbanken zu oder greift sowohl auf Datenbanken als auch auf Ressourcen wie Nachrichten-Middleware zu. Auf diese Weise können wir alle übermittelten oder alle abgebrochenen Transaktionen direkt in mehreren Datenbanken und Nachrichten-Middleware implementieren. Die XA-Spezifikation ist keine Java-Spezifikation, sondern eine universelle Spezifikation. Derzeit unterstützen verschiedene Datenbanken und viele Nachrichten-Middleware die XA-Spezifikation.

JTA ist eine Spezifikation für die Java-Entwicklung, die der XA-Spezifikation entspricht. Wenn wir also sagen, dass wir JTA verwenden, um verteilte Transaktionen zu implementieren, meinen wir eigentlich, JTA-Spezifikationen zu verwenden, um Transaktionen mit mehreren Datenbanken, Nachrichten-Middleware und anderen Ressourcen im System zu implementieren.

Was ist Atomikos?

Atomikos ist ein sehr beliebter Open-Source-Transaktionsmanager und kann in Ihre Spring Boot-Anwendung eingebettet werden. Der Tomcat-Anwendungsserver implementiert die JTA-Spezifikation nicht. Wenn Sie Tomcat als Anwendungsserver verwenden, müssen Sie eine Transaktionsmanagerklasse eines Drittanbieters als globalen Transaktionsmanager verwenden. Dies übernimmt das Atomikos-Framework und integriert die Transaktionsverwaltung in die Anwendung. Hängt nicht vom Anwendungsserver ab.

Spring Boot integriert Atomikos

Es ist sinnlos, über eine Reihe von Theorien zu sprechen. Zeigen Sie mir den Code.

Technologie-Stack: Spring Boot+MyBatis+Atomikos+MySQL

Wenn Sie dem Code in diesem Artikel folgen, achten Sie auf Ihre MySQL-Version.

Erstellen Sie zunächst zwei Datenbanken (my-db_0 und my-db_1) und erstellen Sie dann in jeder Datenbank eine Tabelle.

In Datenbank my-db_0:

CREATE TABLE `t_user_0` (
`id` bigint NOT NULL AUTO_INCREMENT,
`user_name` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
`age` int NOT NULL,
`gender` int NOT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=21 DEFAULT CHARSET=utf8;

In Datenbank my-db_1:

CREATE TABLE `t_user_1` (
`id` bigint NOT NULL AUTO_INCREMENT,
`user_name` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
`age` int NOT NULL,
`gender` int NOT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=14 DEFAULT CHARSET=utf8;

Dies dient nur zur Demonstration verteilter Transaktionen. Machen Sie sich keine Gedanken über die spezifische Bedeutung der Tabelle. Gesamtprojektstruktur

Datenquelle

<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">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.tian</groupId>
    <artifactId>spring-boot-atomikos</artifactId>
    <version>1.0-SNAPSHOT</version>

    <packaging>jar</packaging>

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

    <name>spring-boot-atomikos</name>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <!-- mybatis依赖 -->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>1.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- mysql依赖 -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.16</version>
        </dependency>
        <!--分布式事务-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jta-atomikos</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <!-- 要使生成的jar可运行，需要加入此插件 -->
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
        <resources>
            <resource>
                <directory>src/main/java</directory>
                <excludes>
                    <exclude>**/*.java</exclude>
                </excludes>
            </resource>
            <resource>
                <!-- 编译xml文件 -->
                <directory>src/main/resources</directory>
                <includes>
                    <include>**/*.*</include>
                </includes>
            </resource>
        </resources>
    </build>
</project>

MyBatis Scan

server.port=9001
spring.application.name=atomikos-demo

spring.datasource.user0.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.user0.url=jdbc:mysql://localhost:3306/my-db_0?characterEncoding=utf8&useSSL=false&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true
spring.datasource.user0.user=root
spring.datasource.user0.password=123456

spring.datasource.user1.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.user1.url=jdbc:mysql://localhost:3306/my-db_1?characterEncoding=utf8&useSSL=false&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true
spring.datasource.user1.user=root
spring.datasource.user1.password=123456

mybatis.mapperLocations=classpath:/com/tian/mapper/*/*.xml
mybatis.typeAliasesPackage=com.tian.entity
mybatis.configuration.cache-enabled=true

Der andere ist im Grunde derselbe, das heißt, der Scanpfad wurde geändert in:

/**
 * @author tianwc  公众号：java后端技术全栈、面试专栏
 * @version 1.0.0
 * @date 2023年05月11日 19:38
 * 博客地址：<a href="http://woaijava.cc/">博客地址</a>
 * <p>
 * 配置好两个数据源
 */
@Configuration
public class DataSourceConfig {

    // 将这个对象放入spring容器中（交给Spring管理）
    @Bean
    // 读取 application.yml 中的配置参数映射成为一个对象
    @ConfigurationProperties(prefix = "spring.datasource.user0")
    public XADataSource getDataSource0() {
        // 创建XA连接池
        return new MysqlXADataSource();
    }

    /**
     * 创建Atomikos数据源
     * 注解@DependsOn("druidXADataSourcePre")，在名为druidXADataSourcePre的bean实例化后加载当前bean
     */
    @Bean
    @DependsOn("getDataSource0")
    @Primary
    public DataSource dataSourcePre(@Qualifier("getDataSource0") XADataSource xaDataSource) {
        //这里的AtomikosDataSourceBean使用的是spring提供的
        AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
        atomikosDataSourceBean.setXaDataSource(xaDataSource);
        atomikosDataSourceBean.setMaxPoolSize(20);
        return atomikosDataSourceBean;
    }


    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.user1")
    public XADataSource getDataSource1() {
        // 创建XA连接池
        return new MysqlXADataSource();
    }

    @Bean
    @DependsOn("getDataSource1")
    public DataSource dataSourceSit(@Qualifier("getDataSource1") XADataSource xaDataSource) {
        //这里的AtomikosDataSourceBean使用的是spring提供的
        AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
        atomikosDataSourceBean.setXaDataSource(xaDataSource);
        return atomikosDataSourceBean;
    }
}

mapper.xml

@Configuration
@MapperScan(basePackages = {"com.tian.mapper.user0"}, sqlSessionTemplateRef = "preSqlSessionTemplate")
public class MybatisPreConfig {

    @Autowired
    @Qualifier("dataSourcePre")
    private DataSource dataSource;

    /**
     * 创建 SqlSessionFactory
     */
    @Bean
    @Primary
    public SqlSessionFactory preSqlSessionFactory() throws Exception{
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(dataSource);
        bean.setMapperLocations(new PathMatchingResourcePatternResolver().
                getResources("classpath*:com/tian/mapper/user0/*.xml"));
        return bean.getObject();
    }

    /**
     * 通过 SqlSessionFactory 来创建 SqlSessionTemplate
     */
    @Bean
    @Primary
    public SqlSessionTemplate preSqlSessionTemplate(@Qualifier("preSqlSessionFactory") SqlSessionFactory sqlSessionFactory){
        // SqlSessionTemplate是线程安全的，可以被多个DAO所共享使用
        return new SqlSessionTemplate(sqlSessionFactory);
    }
}

Der andere ist im Grunde derselbe und wird hier veröffentlicht. Die entsprechende Mapper-Schnittstelle ist ebenfalls sehr einfach. Hier ist eine:

("classpath*:com/tian/mapper/user1/*.xml")

Service

<?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="com.tian.mapper.user0.User0Mapper">

    <!-- -->
    <cache eviction="LRU" flushInterval="10000" size="1024"  />

    <resultMap id="BaseResultMap" type="com.tian.entity.User0">
        <id column="id" jdbcType="BIGINT" property="id" />
        <result column="user_name" jdbcType="VARCHAR" property="userName" />
        <result column="age" jdbcType="INTEGER" property="age" />
        <result column="gender" jdbcType="INTEGER" property="gender" />
    </resultMap>

    <sql id="Base_Column_List">
        id, user_name, age, gender
    </sql>
    <insert id="insert" parameterType="com.tian.entity.User0">
        insert into t_user_0 (id, user_name,age, gender)
        values (#{id,jdbcType=BIGINT}, #{userName,jdbcType=VARCHAR},#{age,jdbcType=INTEGER},#{gender,jdbcType=INTEGER})
    </insert>

</mapper>

Controller

public interface User0Mapper {

    int insert(User0 record);
}

Projektstartklasse

/**
 * @author tianwc  公众号：java后端技术全栈、面试专栏
 * @version 1.0.0
 * @date 2023年05月11日 19:38
 * 博客地址：<a href="http://woaijava.cc/">博客地址</a>
 * <p>
 * 模拟三种场景：正常、制造异常、数据库异常
 */
@Service
public class UserServiceImpl implements UserService {

    @Resource
    private User0Mapper user0Mapper;
    @Resource
    private User1Mapper user1Mapper;
    /**
     * 正常逻辑 同时对两个数据库进行 插入数据
     */
    @Transactional
    @Override
    public int transaction1() throws Exception {
        User1 user1 = new User1();
        user1.setUserName("22222");
        user1.setAge(11);
        user1.setGender(0);
        user1Mapper.add(user1);
        System.out.println("---------------------------");
        // sit(数据源1)
        User0 user0 = new User0();
        user0.setUserName("111111");
        user0.setAge(11);
        user0.setGender(0);
        user0Mapper.insert(user0);
        return 1;
    }
    /**
     * 正常逻辑 同时对两个数据库进行 插入数据
     * 数据插入完后  出现异常
     */
    @Transactional
    @Override
    public int transaction2() throws Exception {
        User1 user1 = new User1();
        user1.setUserName("22222");
        user1.setAge(11);
        user1.setGender(0);
        user1Mapper.add(user1);
        System.out.println("---------------------------");
        // sit(数据源1)
        User0 user0 = new User0();
        user0.setUserName("111111");
        user0.setAge(11);
        user0.setGender(0);
        user0Mapper.insert(user0);
        //认为制造一个异常
        int a=1/0;
        return 1;
    }

    /**
     * 第一个数据插入成功  第二个数据插入失败
     */
    @Transactional
    @Override
    public int transaction3() throws Exception {
        User1 user1 = new User1();
        user1.setUserName("22222");
        user1.setAge(11);
        user1.setGender(0);
        user1Mapper.add(user1);
        System.out.println("---------------------------");
        // sit(数据源1)
        User0 user0 = new User0();
       //故意搞长点，模拟插入失败 让前面的数据回滚 user0.setUserName("111110000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001");
        user0.setAge(11);
        user0.setGender(0);
        user0Mapper.insert(user0);
        return 1;
    }
}

Test

Starten Sie das Projekt und testen Sie jeweils die folgenden drei: http://localhost:9001/user/test2 Ergebnis: Es wird eine Ausnahme ausgelöst, dass der Divisor nicht Null sein darf, und es werden keine neuen Daten hinzugefügt entweder Datenbank.

http://localhost:9001/user/test3 Ergebnis: Es wird eine Ausnahme zu einem zu langen Datenfeldwert ausgelöst und es werden keiner Datenbank neue Daten hinzugefügt.

Okay, jetzt haben wir verteilte Transaktionen implementiert.

Das obige ist der detaillierte Inhalt vonSpring Boot+MyBatis+Atomikos+MySQL (mit Quellcode). Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!