美文网首页设计方案k
分布式事务解决方案——可靠消息最终一致性

分布式事务解决方案——可靠消息最终一致性

作者: 小波同学 | 来源:发表于2020-10-17 00:29 被阅读0次

分布式事务理论：分布式事务

一、什么是可靠消息最终一致性事务

可靠消息最终一致性方案是指当事务发起方执行完成本地事务后并发出一条消息，事务参与方(消息消费者)一定能够接收消息并处理事务成功，此方案强调的是只要消息发给事务参与方最终事务要达到一致。

此方案是利用消息中间件完成，如下图：


事务发起方（消息生产方）将消息发给消息中间件，事务参与方从消息中间件接收消息，事务发起方和消息中间件之间，事务参与方（消息消费方）和消息中间件之间都是通过网络通信，由于网络通信的不确定性会导致分布式事务问题。

1、本地事务与消息发送的原子性问题

本地事务与消息发送的原子性问题即：事务发起方在本地事务执行成功后消息必须发出去，否则就丢弃消息。即实现本地事务和消息发送的原子性，要么都成功，要么都失败。本地事务与消息发送的原子性问题是实现可靠消息最终一致性方案的关键问题。

先来尝试下这种操作，先发送消息，再操作数据库：

begin transaction；
    //1.发送MQ
    //2.数据库操作
commit transation;

这种情况下无法保证数据库操作与发送消息的一致性，因为可能发送消息成功，数据库操作失败。

你立马想到第二种方案，先进行数据库操作，再发送消息：

begin transaction；
    //1.数据库操作
    //2.发送MQ
commit transation;

这种情况下貌似没有问题，如果发送MQ消息失败，就会抛出异常，导致数据库事务回滚。但如果是超时异常，数据库回滚，但MQ其实已经正常发送了，同样会导致不一致。

2、事务参与方接收消息的可靠性

事务参与方必须能够从消息队列接收到消息，如果接收消息失败可以重复接收消息。

3、消息重复消费的问题

由于网络的存在，若某一个消费节点超时但是消费成功，此时消息中间件会重复投递此消息，就导致了消息的重复消费。
要解决消息重复消费的问题就要实现事务参与方的方法幂等性。

二、解决方案

1、本地消息表方案

本地消息表这个方案最初是eBay提出的，此方案的核心是通过本地事务保证数据业务操作和消息的一致性，然后通过定时任务将消息发送至消息中间件，待确认消息发送给消费方成功再将消息删除。

下面以注册送积分为例来说明：

下例共有两个微服务交互，用户服务和积分服务，用户服务负责添加用户，积分服务负责增加积分。

交互流程如下：

1、用户注册

用户服务在本地事务新增用户和增加 ”积分消息日志“。（用户表和消息表通过本地事务保证一致）
下边是伪代码

begin transaction；
    //1.新增用户
    //2.存储积分消息日志
commit transation;

这种情况下，本地数据库操作与存储积分消息日志处于同一个事务中，本地数据库操作与记录消息日志操作具备原子性。

2、定时任务扫描日志

如何保证将消息发送给消息队列呢？

经过第一步消息已经写到消息日志表中，可以启动独立的线程，定时对消息日志表中的消息进行扫描并发送至消息中间件，在消息中间件反馈发送成功后删除该消息日志，否则等待定时任务下一周期重试。

3、消费消息

如何保证消费者一定能消费到消息呢？

这里可以使用MQ的ack（即消息确认）机制，消费者监听MQ，如果消费者接收到消息并且业务处理完成后向MQ发送ack（即消息确认），此时说明消费者正常消费消息完成，MQ将不再向消费者推送消息，否则消费者会不断重试向消费者来发送消息。

积分服务接收到”增加积分“消息，开始增加积分，积分增加成功后向消息中间件回应ack，否则消息中间件将重复投递此消息。

由于消息会重复投递，积分服务的”增加积分“功能需要实现幂等性。

2、RocketMQ事务消息方案

RocketMQ 是一个来自阿里巴巴的分布式消息中间件，于 2012 年开源，并在 2017 年正式成为 Apache 顶级项目。据了解，包括阿里云上的消息产品以及收购的子公司在内，阿里集团的消息产品全线都运行在 RocketMQ 之上，并且最近几年的双十一大促中，RocketMQ 都有抢眼表现。Apache RocketMQ 4.3之后的版本正式支持事务消息，为分布式事务实现提供了便利性支持。

RocketMQ 事务消息设计则主要是为了解决 Producer 端的消息发送与本地事务执行的原子性问题，RocketMQ 的设计中 broker 与 producer 端的双向通信能力，使得 broker 天生可以作为一个事务协调者存在；而 RocketMQ本身提供的存储机制为事务消息提供了持久化能力；RocketMQ 的高可用机制以及可靠消息设计则为事务消息在系
统发生异常时依然能够保证达成事务的最终一致性。

在RocketMQ 4.3后实现了完整的事务消息，实际上其实是对本地消息表的一个封装，将本地消息表移动到了MQ内部，解决 Producer 端的消息发送与本地事务执行的原子性问题。

通过冯嘉发布的《RocketMQ 4.3正式发布，支持分布式事务》一文可以看到RocketMQ采用了2PC的方案来提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息。

这张图说明了事务消息的大致方案，分为两个逻辑：正常事务消息的发送及提交、事务消息的补偿流程

事务消息发送及提交：
  • 1、发送消息（half消息）
  • 2、服务端响应消息写入结果
  • 3、根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）
  • 4、根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）
补偿流程：
  • 1、对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”
  • 2、Producer收到回查消息，检查回查消息对应的本地事务的状态
  • 3、根据本地事务状态，重新Commit或者Rollback

补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。

以上RocketMQ事务消息的整体方案，对于了解Notify的同学应该是很熟悉的，下面是之前Notify相关的资料：


执行流程如下：
为方便理解我们还以注册送积分的例子来描述 整个流程。
Producer 即MQ发送方，本例中是用户服务，负责新增用户。MQ订阅方即消息消费方，本例中是积分服务，负责
新增积分。

  • 1、Producer 发送事务消息
    Producer （MQ发送方）发送事务消息至MQ Server，MQ Server将消息状态标记为Prepared（预备状态），注意此时这条消息消费者（MQ订阅方）是无法消费到的。
    本例中，Producer 发送 ”增加积分消息“ 到MQ Server。
  • 2、MQ Server回应消息发送成功
    MQ Server接收到Producer 发送给的消息则回应发送成功表示MQ已接收到消息。
  • 3、Producer 执行本地事务
    Producer 端执行业务代码逻辑，通过本地数据库事务控制。
    本例中，Producer 执行添加用户操作。
  • 4、消息投递
    若Producer 本地事务执行成功则自动向MQServer发送commit消息，MQ Server接收到commit消息后将”增加积分消息“ 状态标记为可消费，此时MQ订阅方（积分服务）即正常消费消息；

若Producer 本地事务执行失败则自动向MQServer发送rollback消息，MQ Server接收到rollback消息后 将删除”增加积分消息“ 。

MQ订阅方（积分服务）消费消息，消费成功则向MQ回应ack，否则将重复接收消息。这里ack默认自动回应，即程序执行正常则自动回应ack。

  • 5、事务回查
    如果执行Producer端本地事务过程中，执行端挂掉，或者超时，MQ Server将会不停的询问同组的其他 Producer来获取事务执行状态，这个过程叫事务回查。MQ Server会根据事务回查结果来决定是否投递消息。

以上主干流程已由RocketMQ实现，对用户侧来说，用户需要分别实现本地事务执行以及本地事务回查方法，因此只需关注本地事务的执行状态即可。

RoacketMQ提供RocketMQLocalTransactionListener接口：

public interface RocketMQLocalTransactionListener {

    /**
    ‐ 发送prepare消息成功此方法被回调，该方法用于执行本地事务
    ‐ @param msg 回传的消息，利用transactionId即可获取到该消息的唯一Id
    ‐ @param arg 调用send方法时传递的参数，当send时候若有额外的参数可以传递到send方法中，这里能获取到
    ‐ @return 返回事务状态，COMMIT：提交 ROLLBACK：回滚 UNKNOW：回调
    */
    RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg);

    /**
    ‐ @param msg 通过获取transactionId来判断这条消息的本地事务执行状态
    ‐ @return 返回事务状态，COMMIT：提交 ROLLBACK：回滚 UNKNOW：回调
    */
    RocketMQLocalTransactionState checkLocalTransaction(Message msg);

}

发送事务消息：
以下是RocketMQ提供用于发送事务消息的API：

TransactionMQProducer producer = new TransactionMQProducer("ProducerGroup");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
//设置TransactionListener实现
producer.setTransactionListener(transactionListener）；
//发送事务消息
SendResult sendResult = producer.sendMessageInTransaction(msg, null);

三、RocketMQ实现可靠消息最终一致性事务

业务说明

本实例通过RocketMQ中间件实现可靠消息最终一致性分布式事务，模拟两个账户的转账交易过程。

两个账户在分别在不同的银行(张三在bank1、李四在bank2)，bank1、bank2是两个微服务。交易过程是，张三给李四转账指定金额。

上述交易步骤，张三扣减金额与给bank2发转账消息，两个操作必须是一个整体性的事务。

程序组成部分

rocketmq 服务端：RocketMQ-4.5.1 release
rocketmq 客户端：RocketMQ-Spring-Boot-starter.2.0.2-RELEASE
微服务框架：
Nacos-Server：1.3.1
SpringBoot：2.2.10.RELEASE
spring-cloud-dependencies：Hoxton.SR8
spring-cloud-alibaba-dependencies：2.2.1.RELEASE

创建数据库

创建bank1库和bank2库，并导入以下表结构和数据(包含张三账户)

DROP TABLE IF EXISTS `account_info`;
CREATE TABLE `account_info` (
    `id` bigint(20) NOT NULL AUTO_INCREMENT,
    `account_name` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '户
    主姓名',
    `account_no` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '银行
    卡号',
    `account_password` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT
    '帐户密码',
    `account_balance` double NULL DEFAULT NULL COMMENT '帐户余额',
    PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;

在bank1、bank2数据库中新增de_duplication，交易记录表(去重表)，用于交易幂等控制。

DROP TABLE IF EXISTS `de_duplication`;

CREATE TABLE `de_duplication` (
    `tx_no` varchar(64) COLLATE utf8_bin NOT NULL,
    `create_time` datetime(0) NULL DEFAULT NULL,
    PRIMARY KEY (`tx_no`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;

启动RocketMQ

Windows系统：
set ROCKETMQ_HOME=[rocketmq服务端解压路径]
start [rocketmq服务端解压路径]/bin/mqnamesrv.cmd

Centos系统：
 进入rocketMQ解压目录下的bin文件夹
nohup sh bin/mqnamesrv & 
日志目录：{rocketMQ解压目录}/logs/rocketmqlogs/namesrv.log

启动broker:

Windows系统：
set ROCKETMQ_HOME=[rocketmq服务端解压路径]
start [rocketmq服务端解压路径]/bin/mqbroker.cmd ‐n 127.0.0.1:9876 autoCreateTopicEnable=true

Centos系统：
进入rocketMQ解压目录下的bin文件夹
nohup sh bin/mqbroker &
日志目录：{rocketMQ解压目录}/logs/rocketmqlogs/broker.log

创建工程

  • 1)、工程引入maven依赖
<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.10.RELEASE</version>
    <relativePath/>
</parent>

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
</properties>

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

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

    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        <exclusions>
            <exclusion>
                <groupId>com.alibaba.nacos</groupId>
                <artifactId>nacos-client</artifactId>
            </exclusion>
        </exclusions>
    </dependency>

    <dependency>
        <groupId>com.alibaba.nacos</groupId>
        <artifactId>nacos-client</artifactId>
        <version>1.3.1</version>
    </dependency>

    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.0.3</version>
    </dependency>

    <dependency>
        <groupId>tk.mybatis</groupId>
        <artifactId>mapper-spring-boot-starter</artifactId>
        <version>2.1.5</version>
    </dependency>

    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.18</version>
    </dependency>

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.12</version>
        <scope>provided</scope>
    </dependency>

    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.74</version>
    </dependency>
</dependencies>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>Hoxton.SR8</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>

        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-alibaba-dependencies</artifactId>
            <version>2.2.1.RELEASE</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
        </plugin>

        <plugin>
            <groupId>org.mybatis.generator</groupId>
            <artifactId>mybatis-generator-maven-plugin</artifactId>
            <version>1.3.6</version>
            <configuration>
                <configurationFile>
                    ${basedir}/src/main/resources/generator/generatorConfig.xml
                </configurationFile>
                <overwrite>true</overwrite>
                <verbose>true</verbose>
            </configuration>
            <dependencies>
                <dependency>
                    <groupId>mysql</groupId>
                    <artifactId>mysql-connector-java</artifactId>
                    <version>5.1.41</version>
                </dependency>
                <dependency>
                    <groupId>tk.mybatis</groupId>
                    <artifactId>mapper</artifactId>
                    <version>4.1.5</version>
                </dependency>
            </dependencies>
        </plugin>
    </plugins>
</build>
Producer端
  • 1)、application.properties
server.port=8090

spring.datasource.url=jdbc:mysql://localhost:3306/trade_bank1?useUnicode=true&useAffectedRows=true&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=UTC
spring.datasource.username=root
spring.datasource.password=yibo
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver

rocketmq.producer.group = producer_bank1
rocketmq.name-server = 127.0.0.1:9876

logging.level.root = info
logging.level.org.springframework.web = info
logging.level.cn.itcast.wanxintx.ensuredemo  = debug
  • 2)、Controller
@RestController
@Slf4j
@RequestMapping("/rocket1")
public class AccountInfoController {

    @Autowired
    private AccountInfoService accountInfoService;

    @GetMapping(value = "/transfer")
    public String transfer(@RequestParam("accountNo")String accountNo, @RequestParam("amount") Double amount){
        //创建一个事务id，作为消息内容发到mq
        String tx_no = UUID.randomUUID().toString();
        AccountChangeEvent accountChangeEvent = new AccountChangeEvent(accountNo,amount,tx_no);
        //发送消息
        accountInfoService.sendUpdateAccountBalance(accountChangeEvent);
        return "转账成功";
    }
}
  • 3)、Service
@Service
@Slf4j
public class AccountInfoService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private AccountInfoMapper accountInfoMapper;

    /**
     * 向mq发送转账消息
     * @param accountChangeEvent
     */
    public void sendUpdateAccountBalance(AccountChangeEvent accountChangeEvent){
        //将accountChangeEvent转成json
        JSONObject jsonObject =new JSONObject();
        jsonObject.put("accountChange",accountChangeEvent);
        String jsonString = jsonObject.toJSONString();
        //生成message类型
        Message<String> message = MessageBuilder.withPayload(jsonString).build();
        //发送一条事务消息
        /**
         * String txProducerGroup 生产组
         * String destination topic，
         * Message<?> message, 消息内容
         * Object arg 参数
         */
        rocketMQTemplate.sendMessageInTransaction("producer_group_txmsg_bank1","topic_txmsg",message,null);
    }

    /**
     * 更新账户，扣减金额
     * @param accountChangeEvent
     */
    @Transactional
    public void doUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
        //幂等判断
        if(accountInfoMapper.isExistTx(accountChangeEvent.getTxNo())>0){
            return ;
        }
        //扣减金额
        accountInfoMapper.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount() * -1);
        //添加事务日志
        accountInfoMapper.addTx(accountChangeEvent.getTxNo());
    }
}
  • 4)、RocketMQLocalTransactionListener
    编写RocketMQLocalTransactionListener接口实现类，实现执行本地事务和事务回查两个方法。
@Component
@RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg_bank1")
@Slf4j
public class ProducerTransactionListener implements RocketMQLocalTransactionListener {

    @Autowired
    private AccountInfoService accountInfoService;

    @Autowired
    private AccountInfoMapper accountInfoMapper;

    /**
     * 事务消息发送后的回调方法，当消息发送给mq成功，此方法被回调
     * @param message
     * @param o
     * @return
     */
    @Transactional
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        try {
            //解析message，转成AccountChangeEvent
            String messageString = new String((byte[]) message.getPayload());
            JSONObject jsonObject = JSONObject.parseObject(messageString);
            String accountChangeString = jsonObject.getString("accountChange");
            //将accountChange（json）转成AccountChangeEvent
            AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
            //执行本地事务，扣减金额
            accountInfoService.doUpdateAccountBalance(accountChangeEvent);
            //当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit消息，mq将消息的状态改为可消费
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("ProducerTransactionListener execute Exception",e);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 事务状态回查，查询是否扣减金额
     * @param message
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        //解析message，转成AccountChangeEvent
        String messageString = new String((byte[]) message.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        String accountChangeString = jsonObject.getString("accountChange");
        //将accountChange（json）转成AccountChangeEvent
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
        //事务id
        String txNo = accountChangeEvent.getTxNo();
        int existTx = accountInfoMapper.isExistTx(txNo);
        if(existTx>0){
            return RocketMQLocalTransactionState.COMMIT;
        }else{
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}
  • 5)、Mapper
public interface AccountInfoMapper extends Mapper<AccountInfo> {

    int updateAccountBalance(@Param("accountNo") String accountNo, @Param("amount") Double amount);

    AccountInfo findByIdAccountNo(@Param("accountNo") String accountNo);

    int isExistTx(String txNo);

    int addTx(String txNo);
}

<mapper namespace="com.yibo.rocketmq.mapper.AccountInfoMapper">
  <resultMap id="BaseResultMap" type="com.yibo.rocketmq.domain.entity.AccountInfo">
    <!--
      WARNING - @mbg.generated
    -->
    <id column="id" jdbcType="BIGINT" property="id" />
    <result column="account_name" jdbcType="VARCHAR" property="accountName" />
    <result column="account_no" jdbcType="VARCHAR" property="accountNo" />
    <result column="account_password" jdbcType="VARCHAR" property="accountPassword" />
    <result column="account_balance" jdbcType="BIGINT" property="accountBalance" />
  </resultMap>

  <update id="updateAccountBalance">
    update account_info set account_balance=account_balance+#{amount} where account_no=#{accountNo}
  </update>

  <select id="findByIdAccountNo">
    select * from account_info where where account_no=#{accountNo}
  </select>

  <select id="isExistTx">
    select count(1) from de_duplication where tx_no = #{txNo}
  </select>

  <insert id="addTx">
    insert into de_duplication values(#{txNo},now())
  </insert>

</mapper>
Consumer端

  • 1、监听MQ，接收消息。

  • 2、接收到消息增加账户金额。

  • 1)、application.properties

server.port=8092

spring.datasource.url=jdbc:mysql://localhost:3306/trade_bank2?useUnicode=true&useAffectedRows=true&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=UTC
spring.datasource.username=root
spring.datasource.password=yibo
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver

rocketmq.producer.group = producer_bank2
rocketmq.name-server = 127.0.0.1:9876

logging.level.root = info
logging.level.org.springframework.web = info
logging.level.cn.itcast.wanxintx.ensuredemo  = debug
  • 2)、RocketMQListener
@Component
@Slf4j
@RocketMQMessageListener(consumerGroup = "consumer_group_txmsg_bank2",topic = "topic_txmsg")
public class TransactionConsumer implements RocketMQListener<AccountChangeEvent> {

    @Autowired
    private AccountInfoService accountInfoService;

    @Override
    public void onMessage(AccountChangeEvent accountChangeEvent) {
        log.info("开始消费消息:{}", JSON.toJSONString(accountChangeEvent));
        //设置账号为李四的
        accountChangeEvent.setAccountNo("2");
        //更新本地账户，增加金额
        accountInfoService.addAccountInfoBalance(accountChangeEvent);
    }
}
  • 3)、Service
@Service
@Slf4j
public class AccountInfoService {

    @Autowired
    private AccountInfoMapper accountInfoMapper;

    /**
     * 更新账户，增加金额
     * @param accountChangeEvent
     */
    @Transactional
    public void addAccountInfoBalance(AccountChangeEvent accountChangeEvent) {
        log.info("bank2更新本地账号，账号：{},金额：{}",accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());
        if(accountInfoMapper.isExistTx(accountChangeEvent.getTxNo())>0){
            return ;
        }

        //增加金额
        accountInfoMapper.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());

        //添加事务记录，用于幂等
        accountInfoMapper.addTx(accountChangeEvent.getTxNo());
    }
}
  • 4)、Mapper
public interface AccountInfoMapper extends Mapper<AccountInfo> {

    int updateAccountBalance(@Param("accountNo") String accountNo, @Param("amount") Double amount);

    int isExistTx(String txNo);

    int addTx(String txNo);
}

<mapper namespace="com.yibo.rocketmq.mapper.AccountInfoMapper">
  <resultMap id="BaseResultMap" type="com.yibo.rocketmq.domain.entity.AccountInfo">
    <!--
      WARNING - @mbg.generated
    -->
    <id column="id" jdbcType="BIGINT" property="id" />
    <result column="account_name" jdbcType="VARCHAR" property="accountName" />
    <result column="account_no" jdbcType="VARCHAR" property="accountNo" />
    <result column="account_password" jdbcType="VARCHAR" property="accountPassword" />
    <result column="account_balance" jdbcType="BIGINT" property="accountBalance" />
  </resultMap>


  <update id="updateAccountBalance">
    update account_info set account_balance=account_balance+#{amount} where account_no=#{accountNo}
  </update>

  <select id="isExistTx">
    select count(1) from de_duplication where tx_no = #{txNo}
  </select>

  <insert id="addTx">
    insert into de_duplication values(#{txNo},now())
  </insert>
</mapper>

总结

可靠消息最终一致性就是保证消息从生产方经过消息中间件传递到消费方的一致性，本案例使用了RocketMQ作为消息中间件，RocketMQ主要解决了两个功能：

  • 1、本地事务与消息发送的原子性问题。
  • 2、事务参与方接收消息的可靠性。

可靠消息最终一致性事务适合执行周期长且实时性要求不高的场景。引入消息机制后，同步的事务操作变为基于消息执行的异步操作, 避免了分布式事务中的同步阻塞操作的影响，并实现了两个服务的解耦。

github源码地址：https://github.com/jjhyb/distributed-transaction

参考：
https://www.cnblogs.com/zeussbook/p/11798658.html

https://www.pianshen.com/article/54181331893/

https://www.cnblogs.com/hzmark/p/rocket_txn.html

相关文章

网友评论

    本文标题：分布式事务解决方案——可靠消息最终一致性

    本文链接：https://www.haomeiwen.com/subject/hzempktx.html

    延伸阅读

    深度阅读

    • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

    栏目导航

    热点阅读

    设计方案

    k

    关于我们|服务条款|联系我们|分布式事务解决方案——可靠消息最终一致性|投稿指南|网站地图|RSS订阅|排版工具|手机版

    提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

    Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

    备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

    本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

    所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！