點擊上方藍色字體，選擇“標(biāo)星公眾號”

優(yōu)質(zhì)文章，第一時間送達

作者 | C紫楓

來源 | urlify.cn/BjYv6f

分布式事務(wù)問題

一次業(yè)務(wù)操作需要垮多個數(shù)據(jù)源或需要垮多個系統(tǒng)進行遠程調(diào)用,就會產(chǎn)生分布式事務(wù)問題。

Seata簡介

是什么？

Seata是一款開源的分布式事務(wù)解決方案,致力于在微服務(wù)架構(gòu)下提供高性能和簡單易用的分布式事務(wù)服務(wù)。
官網(wǎng)地址：http://seata.io/zh-cn/

一個典型的分布式事務(wù)過程

分布式事務(wù)處理過程:XID+三組件模型

XID

Transaction ID(XID):全局唯一的事務(wù)id

三組件概念

Transaction Coordinator(TC):事務(wù)協(xié)調(diào)器,維護全局事務(wù)的運行狀態(tài),負責(zé)協(xié)調(diào)并驅(qū)動全局事務(wù)的提交或回滾。
Transaction Manager(TM)：控制全局事務(wù)的邊界,負責(zé)開啟一個全局事務(wù),并最終發(fā)起全局提交或全局回滾的決議。
Resource Manager(RM)：控制分支事務(wù),負責(zé)分支注冊、狀態(tài)匯報,并接受事務(wù)協(xié)調(diào)的指令,驅(qū)動分支(本地)事務(wù)的提交和回滾。

下載地址

發(fā)布說明: https://github.com/seata/seata/releases

怎么使用

本地@Transational
全局@GlobalTranstional
seata的分布式交易解決方案：

配置說明

seata-server-0.9.0.zip解壓到指定目錄并修改conf目錄下的file.conf配置文件，主要修改的內(nèi)容：自定義事務(wù)組名稱+事務(wù)日志存儲模式為db+數(shù)據(jù)庫連接

mysql5.7數(shù)據(jù)庫新建庫seata

建表db_store.sql在seata-server-0.9.0\seata\conf目錄里面db_store.sql（注意seata1.0開始就沒這個數(shù)據(jù)庫文件了）

-- the table to store GlobalSession data
drop table if exists `global_table`;
create table `global_table` (
  `xid` varchar(128)  not null,
  `transaction_id` bigint,
  `status` tinyint not null,
  `application_id` varchar(32),
  `transaction_service_group` varchar(32),
  `transaction_name` varchar(128),
  `timeout` int,
  `begin_time` bigint,
  `application_data` varchar(2000),
  `gmt_create` datetime,
  `gmt_modified` datetime,
  primary key (`xid`),
  key `idx_gmt_modified_status` (`gmt_modified`, `status`),
  key `idx_transaction_id` (`transaction_id`)
);
 
-- the table to store BranchSession data
drop table if exists `branch_table`;
create table `branch_table` (
  `branch_id` bigint not null,
  `xid` varchar(128) not null,
  `transaction_id` bigint ,
  `resource_group_id` varchar(32),
  `resource_id` varchar(256) ,
  `lock_key` varchar(128) ,
  `branch_type` varchar(8) ,
  `status` tinyint,
  `client_id` varchar(64),
  `application_data` varchar(2000),
  `gmt_create` datetime,
  `gmt_modified` datetime,
  primary key (`branch_id`),
  key `idx_xid` (`xid`)
);
 
-- the table to store lock data
drop table if exists `lock_table`;
create table `lock_table` (
  `row_key` varchar(128) not null,
  `xid` varchar(96),
  `transaction_id` long ,
  `branch_id` long,
  `resource_id` varchar(256) ,
  `table_name` varchar(32) ,
  `pk` varchar(36) ,
  `gmt_create` datetime ,
  `gmt_modified` datetime,
  primary key(`row_key`)
);

修改seata-server-0.9.0\seata\conf目錄下的registry.conf目錄下的registry.conf配置文件

先啟動Nacos端口號8848

再啟動seata-server

seata-server-0.9.0\seata\bin --->seata-server.bat
遇到的問題：啟動閃退報jvm內(nèi)存不夠（可能是我用的是openJdk）
解決方案：https://blog.csdn.net/weixin_39724194/article/details/107329330

功能模擬

訂單/庫存/賬戶業(yè)務(wù)數(shù)據(jù)庫準(zhǔn)備

創(chuàng)建業(yè)務(wù)數(shù)據(jù)庫

seata_order:存儲訂單的數(shù)據(jù)庫
seata_storage:存儲庫存的數(shù)據(jù)庫
seata_account:存儲賬戶信息的數(shù)據(jù)庫

按照上述3庫分別建立對應(yīng)業(yè)務(wù)表

seata_order庫下新建t_order表

DROP TABLE IF EXISTS `t_order`;
CREATE TABLE `t_order`  (
  `id` bigint(11) NOT NULL AUTO_INCREMENT,
  `user_id` bigint(20) DEFAULT NULL COMMENT '用戶id',
  `product_id` bigint(11) DEFAULT NULL COMMENT '產(chǎn)品id',
  `count` int(11) DEFAULT NULL COMMENT '數(shù)量',
  `money` decimal(11, 0) DEFAULT NULL COMMENT '金額',
  `status` int(1) DEFAULT NULL COMMENT '訂單狀態(tài):  0:創(chuàng)建中 1:已完結(jié)',
  PRIMARY KEY (`int`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '訂單表' ROW_FORMAT = Dynamic;

seata_storage庫下新建t_storage表

DROP TABLE IF EXISTS `t_storage`;
CREATE TABLE `t_storage`  (
  `id` bigint(11) NOT NULL AUTO_INCREMENT,
  `product_id` bigint(11) DEFAULT NULL COMMENT '產(chǎn)品id',
  `total` int(11) DEFAULT NULL COMMENT '總庫存',
  `used` int(11) DEFAULT NULL COMMENT '已用庫存',
  `residue` int(11) DEFAULT NULL COMMENT '剩余庫存',
  PRIMARY KEY (`int`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '庫存' ROW_FORMAT = Dynamic;
INSERT INTO `t_storage` VALUES (1, 1, 100, 0, 100);

seata_account庫下新建t_account表

CREATE TABLE `t_account`  (
  `id` bigint(11) NOT NULL COMMENT 'id',
  `user_id` bigint(11) DEFAULT NULL COMMENT '用戶id',
  `total` decimal(10, 0) DEFAULT NULL COMMENT '總額度',
  `used` decimal(10, 0) DEFAULT NULL COMMENT '已用余額',
  `residue` decimal(10, 0) DEFAULT NULL COMMENT '剩余可用額度',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '賬戶表' ROW_FORMAT = Dynamic;
 
INSERT INTO `t_account` VALUES (1, 1, 1000, 0, 1000);

按照上述3庫分別建立對應(yīng)的回滾日志表

訂單-庫存-賬戶3個庫下都需要建各自獨立的回滾日志表,seata-server-0.9.0\seata\conf\目錄下的db_undo_log.sql,建表SQL:

CREATE TABLE `undo_log` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `branch_id` bigint(20) NOT NULL,
  `xid` varchar(100) NOT NULL,
  `context` varchar(128) NOT NULL,
  `rollback_info` longblob NOT NULL,
  `log_status` int(11) NOT NULL,
  `log_created` datetime NOT NULL,
  `log_modified` datetime NOT NULL,
  `ext` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

最終效果

訂單/庫存/賬戶業(yè)務(wù)微服務(wù)準(zhǔn)備

業(yè)務(wù)需求：下訂單->減庫存->扣余額->改(訂單)狀態(tài)

新建訂單Order-Module

1.seata-order-service2001
2.POM

  <!--seata-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
            <exclusions>
                <exclusion>
                    <artifactId>seata-all</artifactId>
                    <groupId>io.seata</groupId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>io.seata</groupId>
            <artifactId>seata-all</artifactId>
            <version>1.0.0</version>
        </dependency>

3.YML

server:
  port: 2001

spring:
  application:
    name: seata-order-service
  cloud:
    alibaba:
      seata:
        #自定義事務(wù)組名稱需要與seata-server中的對應(yīng)，在file.conf中的service中
        tx-service-group: fsp_tx_group
    nacos:
      discovery:
        server-addr: localhost:8848
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://localhost:3306/seata_order
    username: root
    password: root

feign:
  hystrix:
    enabled: false

logging:
  level:
    io:
      seata: info

mybatis:
  mapperLocations: classpath:mapper/*.xml

4.file.conf
拷貝seata-server/conf目錄下的file.conf
5.registry.conf
拷貝seata-server/conf目錄下的registry.conf
6.主啟動類

package com.czf.springcloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;

@EnableDiscoveryClient
@EnableFeignClients
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)//取消數(shù)據(jù)源的自動創(chuàng)建
public class SeataOrderMainApp2001
{

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

7.Config配置

package com.czf.springcloud.config;

import com.alibaba.druid.pool.DruidDataSource;
import io.seata.rm.datasource.DataSourceProxy;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.transaction.SpringManagedTransactionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;

import javax.sql.DataSource;

/**
 * @auther zzyy
 * @create 2019-12-11 16:58
 * 使用Seata對數(shù)據(jù)源進行代理
 */
@Configuration
public class DataSourceProxyConfig {

    @Value("${mybatis.mapperLocations}")
    private String mapperLocations;

    @Bean
    @ConfigurationProperties(prefix = "spring.datasource")
    public DataSource druidDataSource(){
        return new DruidDataSource();
    }

    @Bean
    public DataSourceProxy dataSourceProxy(DataSource dataSource) {
        return new DataSourceProxy(dataSource);
    }

    @Bean
    public SqlSessionFactory sqlSessionFactoryBean(DataSourceProxy dataSourceProxy) throws Exception {
        SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
        sqlSessionFactoryBean.setDataSource(dataSourceProxy);//注入數(shù)據(jù)源
        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources(mapperLocations));//相當(dāng)于mapperScaner掃描器
        sqlSessionFactoryBean.setTransactionFactory(new SpringManagedTransactionFactory());//配置事務(wù)
        return sqlSessionFactoryBean.getObject();
    }

}

package com.czf.springcloud.config;

import org.mybatis.spring.annotation.MapperScan;
import org.springframework.context.annotation.Configuration;

/**
 * @auther zzyy
 * @create 2019-12-11 16:57
 */
@Configuration
@MapperScan({"com.czf.springcloud.dao"})
public class MyBatisConfig {
}

8.Service接口及實現(xiàn)
OrderService

package com.czf.springcloud.service;
import com.czf.springcloud.domain.Order;

/**
 * @auther zzyy
 * @create 2020-02-26 15:19
 */
public interface OrderService
{
    void create(Order order);
}

OrderServiceIplm

package com.czf.springcloud.service.impl;
import com.czf.springcloud.dao.OrderDao;
import com.czf.springcloud.domain.Order;
import com.czf.springcloud.service.AccountService;
import com.czf.springcloud.service.OrderService;
import com.czf.springcloud.service.StorageService;
import io.seata.spring.annotation.GlobalTransactional;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;

/**
 */
@Service
@Slf4j
public class OrderServiceImpl implements OrderService
{
    @Resource
    private OrderDao orderDao;
    @Resource
    private StorageService storageService;
    @Resource
    private AccountService accountService;

    /**
     * 創(chuàng)建訂單->調(diào)用庫存服務(wù)扣減庫存->調(diào)用賬戶服務(wù)扣減賬戶余額->修改訂單狀態(tài)
     * 簡單說：下訂單->扣庫存->減余額->改狀態(tài)
     */
    @Override
    @GlobalTransactional(name = "fsp-create-order",rollbackFor = Exception.class)
    public void create(Order order)
    {
        log.info("----->開始新建訂單");
        //1 新建訂單
        orderDao.create(order);

        //2 扣減庫存
        log.info("----->訂單微服務(wù)開始調(diào)用庫存，做扣減Count");
        storageService.decrease(order.getProductId(),order.getCount());
        log.info("----->訂單微服務(wù)開始調(diào)用庫存，做扣減end");

        //3 扣減賬戶
        log.info("----->訂單微服務(wù)開始調(diào)用賬戶，做扣減Money");
        accountService.decrease(order.getUserId(),order.getMoney());
        log.info("----->訂單微服務(wù)開始調(diào)用賬戶，做扣減end");

        //4 修改訂單狀態(tài)，從零到1,1代表已經(jīng)完成
        log.info("----->修改訂單狀態(tài)開始");
        orderDao.update(order.getUserId(),0);
        log.info("----->修改訂單狀態(tài)結(jié)束");

        log.info("----->下訂單結(jié)束了，O(∩_∩)O哈哈~");

    }
}

AccountService

@FeignClient(value = "seata-account-service")//調(diào)用2003的微服務(wù)-----服務(wù)調(diào)用
public interface AccountService
{
    @PostMapping(value = "/account/decrease")
    CommonResult decrease(@RequestParam("userId") Long userId, @RequestParam("money") BigDecimal money);
}

StorageService

@FeignClient(value = "seata-storage-service")//調(diào)用2002的微服務(wù)
public interface StorageService
{
    @PostMapping(value = "/storage/decrease")
    CommonResult decrease(@RequestParam("productId") Long productId, @RequestParam("count") Integer count);
}

9.Controller

@RestController
public class OrderController
{
    @Resource
    private OrderService orderService;


    @GetMapping("/order/create")
    public CommonResult create(Order order)
    {
        orderService.create(order);
        return new CommonResult(200,"訂單創(chuàng)建成功");
    }
}

新建庫存Storage-Module(seata-storage-service2002)同理

controller

@RestController
public class AccountController {

    @Resource
    AccountService accountService;

    /**
     * 扣減賬戶余額
     */
    @RequestMapping("/account/decrease")
    public CommonResult decrease(@RequestParam("userId") Long userId, @RequestParam("money") BigDecimal money){
        accountService.decrease(userId,money);
        return new CommonResult(200,"扣減賬戶余額成功！");
    }
}

service

public interface AccountService {

    /**
     * 扣減賬戶余額
     * @param userId 用戶id
     * @param money 金額
     */
    void decrease(@RequestParam("userId") Long userId, @RequestParam("money") BigDecimal money);

/**
 * 賬戶業(yè)務(wù)實現(xiàn)類
 * Created by zzyy on 2019/11/11.
 */
@Service
public class AccountServiceImpl implements AccountService {

    private static final Logger LOGGER = LoggerFactory.getLogger(AccountServiceImpl.class);


    @Resource
    AccountDao accountDao;

    /**
     * 扣減賬戶余額
     */
    @Override
    public void decrease(Long userId, BigDecimal money) {
        LOGGER.info("------->account-service中扣減賬戶余額開始");
        //模擬超時異常，全局事務(wù)回滾
        //暫停幾秒鐘線程
        try { TimeUnit.SECONDS.sleep(20); } catch (InterruptedException e) { e.printStackTrace(); }
        accountDao.decrease(userId,money);
        LOGGER.info("------->account-service中扣減賬戶余額結(jié)束");
    }
}

新建賬戶Account-Module(seata-account-service2003)