使用Seata完全解決Spring Cloud中的分佈式事務問題！

時間  2019-11-21
標籤 java node mysql git github redis spring sql 數據庫 bash 欄目 Spring 简体版
原文   原文鏈接

SpringBoot實戰電商項目mall（25k+star）地址：github.com/macrozheng/…java

摘要

Seata是Alibaba開源的一款分佈式事務解決方案，致力於提供高性能和簡單易用的分佈式事務服務，本文將經過一個簡單的下單業務場景來對其用法進行詳細介紹。node

什麼是分佈式事務問題？

單體應用

單體應用中，一個業務操做須要調用三個模塊完成，此時數據的一致性由本地事務來保證。mysql

微服務應用

隨着業務需求的變化，單體應用被拆分紅微服務應用，原來的三個模塊被拆分紅三個獨立的應用，分別使用獨立的數據源，業務操做須要調用三個服務來完成。此時每一個服務內部的數據一致性由本地事務來保證，可是全局的數據一致性問題無法保證。git

小結

在微服務架構中因爲全局數據一致性無法保證產生的問題就是分佈式事務問題。簡單來講，一次業務操做須要操做多個數據源或須要進行遠程調用，就會產生分佈式事務問題。github

Seata簡介

Seata 是一款開源的分佈式事務解決方案，致力於提供高性能和簡單易用的分佈式事務服務。Seata 將爲用戶提供了 AT、TCC、SAGA 和 XA 事務模式，爲用戶打造一站式的分佈式解決方案。redis

Seata原理和設計

定義一個分佈式事務

咱們能夠把一個分佈式事務理解成一個包含了若干分支事務的全局事務，全局事務的職責是協調其下管轄的分支事務達成一致，要麼一塊兒成功提交，要麼一塊兒失敗回滾。此外，一般分支事務自己就是一個知足ACID的本地事務。這是咱們對分佈式事務結構的基本認識，與 XA 是一致的。spring

協議分佈式事務處理過程的三個組件

  • Transaction Coordinator (TC)： 事務協調器，維護全局事務的運行狀態，負責協調並驅動全局事務的提交或回滾；
  • Transaction Manager (TM)： 控制全局事務的邊界，負責開啓一個全局事務，並最終發起全局提交或全局回滾的決議；
  • Resource Manager (RM)： 控制分支事務，負責分支註冊、狀態彙報，並接收事務協調器的指令，驅動分支（本地）事務的提交和回滾。

一個典型的分佈式事務過程

  • TM 向 TC 申請開啓一個全局事務，全局事務建立成功並生成一個全局惟一的 XID；
  • XID 在微服務調用鏈路的上下文中傳播；
  • RM 向 TC 註冊分支事務，將其歸入 XID 對應全局事務的管轄；
  • TM 向 TC 發起針對 XID 的全局提交或回滾決議；
  • TC 調度 XID 下管轄的所有分支事務完成提交或回滾請求。

seata-server的安裝與配置

  • 咱們先從官網下載seata-server，這裏下載的是seata-server-0.9.0.zip，下載地址：github.com/seata/seata…sql

  • 這裏咱們使用Nacos做爲註冊中心，Nacos的安裝及使用能夠參考：Spring Cloud Alibaba：Nacos 做爲註冊中心和配置中心使用；數據庫

  • 解壓seata-server安裝包到指定目錄，修改conf目錄下的file.conf配置文件，主要修改自定義事務組名稱，事務日誌存儲模式爲db及數據庫鏈接信息；bash

service {
  #vgroup->rgroup
  vgroup_mapping.fsp_tx_group = "default" #修改事務組名稱爲：fsp_tx_group，和客戶端自定義的名稱對應
  #only support single node
  default.grouplist = "127.0.0.1:8091"
  #degrade current not support
  enableDegrade = false
  #disable
  disable = false
  #unit ms,s,m,h,d represents milliseconds, seconds, minutes, hours, days, default permanent
  max.commit.retry.timeout = "-1"
  max.rollback.retry.timeout = "-1"
}

## transaction log store
store {
  ## store mode: file、db
  mode = "db" #修改此處將事務信息存儲到數據庫中

  ## database store
  db {
    ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
    datasource = "dbcp"
    ## mysql/oracle/h2/oceanbase etc.
    db-type = "mysql"
    driver-class-name = "com.mysql.jdbc.Driver"
    url = "jdbc:mysql://localhost:3306/seat-server" #修改數據庫鏈接地址
    user = "root" #修改數據庫用戶名
    password = "root" #修改數據庫密碼
    min-conn = 1
    max-conn = 3
    global.table = "global_table"
    branch.table = "branch_table"
    lock-table = "lock_table"
    query-limit = 100
  }
}
複製代碼

  • 因爲咱們使用了db模式存儲事務日誌，因此咱們須要建立一個seat-server數據庫，建表sql在seata-server的/conf/db_store.sql中；

  • 修改conf目錄下的registry.conf配置文件，指明註冊中心爲nacos，及修改nacos鏈接信息便可；

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd三、sofa
  type = "nacos" #改成nacos

  nacos {
    serverAddr = "localhost:8848" #改成nacos的鏈接地址
    namespace = ""
    cluster = "default"
  }
}

複製代碼
  • 先啓動Nacos，再使用seata-server中/bin/seata-server.bat文件啓動seata-server。

數據庫準備

建立業務數據庫

  • seat-order：存儲訂單的數據庫；
  • seat-storage：存儲庫存的數據庫；
  • seat-account：存儲帳戶信息的數據庫。

初始化業務表

order表

CREATE TABLE `order` (
  `id` bigint(11) NOT NULL AUTO_INCREMENT,
  `user_id` bigint(11) DEFAULT NULL COMMENT '用戶id',
  `product_id` bigint(11) DEFAULT NULL COMMENT '產品id',
  `count` int(11) DEFAULT NULL COMMENT '數量',
  `money` decimal(11,0) DEFAULT NULL COMMENT '金額',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8;

ALTER TABLE `order` ADD COLUMN `status` int(1) DEFAULT NULL COMMENT '訂單狀態：0：建立中；1：已完結' AFTER `money` ;
複製代碼

storage表

CREATE TABLE `storage` (
                         `id` bigint(11) NOT NULL AUTO_INCREMENT,
                         `product_id` bigint(11) DEFAULT NULL COMMENT '產品id',
                         `total` int(11) DEFAULT NULL COMMENT '總庫存',
                         `used` int(11) DEFAULT NULL COMMENT '已用庫存',
                         `residue` int(11) DEFAULT NULL COMMENT '剩餘庫存',
                         PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

INSERT INTO `seat-storage`.`storage` (`id`, `product_id`, `total`, `used`, `residue`) VALUES ('1', '1', '100', '0', '100');

複製代碼

account表

CREATE TABLE `account` (
  `id` bigint(11) NOT NULL AUTO_INCREMENT 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 '0' COMMENT '剩餘可用額度',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

INSERT INTO `seat-account`.`account` (`id`, `user_id`, `total`, `used`, `residue`) VALUES ('1', '1', '1000', '0', '1000');

複製代碼

建立日誌回滾表

使用Seata還須要在每一個數據庫中建立日誌表，建表sql在seata-server的/conf/db_undo_log.sql中。

完整數據庫示意圖

製造一個分佈式事務問題

這裏咱們會建立三個服務，一個訂單服務，一個庫存服務，一個帳戶服務。當用戶下單時，會在訂單服務中建立一個訂單，而後經過遠程調用庫存服務來扣減下單商品的庫存，再經過遠程調用帳戶服務來扣減用戶帳戶裏面的餘額，最後在訂單服務中修改訂單狀態爲已完成。該操做跨越三個數據庫，有兩次遠程調用，很明顯會有分佈式事務問題。

客戶端配置

  • 對seata-order-service、seata-storage-service和seata-account-service三個seata的客戶端進行配置，它們配置大體相同，咱們下面以seata-order-service的配置爲例；

  • 修改application.yml文件，自定義事務組的名稱；

spring:
 cloud:
 alibaba:
 seata:
 tx-service-group: fsp_tx_group #自定義事務組名稱須要與seata-server中的對應
複製代碼
  • 添加並修改file.conf配置文件，主要是修改自定義事務組名稱；
service {
  #vgroup->rgroup
  vgroup_mapping.fsp_tx_group = "default" #修改自定義事務組名稱
  #only support single node
  default.grouplist = "127.0.0.1:8091"
  #degrade current not support
  enableDegrade = false
  #disable
  disable = false
  #unit ms,s,m,h,d represents milliseconds, seconds, minutes, hours, days, default permanent
  max.commit.retry.timeout = "-1"
  max.rollback.retry.timeout = "-1"
  disableGlobalTransaction = false
}
複製代碼
  • 添加並修改registry.conf配置文件，主要是將註冊中心改成nacos；
registry {
  # file 、nacos 、eureka、redis、zk
  type = "nacos" #修改成nacos

  nacos {
    serverAddr = "localhost:8848" #修改成nacos的鏈接地址
    namespace = ""
    cluster = "default"
  }
}

複製代碼
  • 在啓動類中取消數據源的自動建立：
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
@EnableDiscoveryClient
@EnableFeignClients
public class SeataOrderServiceApplication {

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

}
複製代碼
  • 建立配置使用Seata對數據源進行代理：
/** * 使用Seata對數據源進行代理 * Created by macro on 2019/11/11. */
@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);
        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver()
                .getResources(mapperLocations));
        sqlSessionFactoryBean.setTransactionFactory(new SpringManagedTransactionFactory());
        return sqlSessionFactoryBean.getObject();
    }

}
複製代碼
  • 使用@GlobalTransactional註解開啓分佈式事務：
package com.macro.cloud.service.impl;

import com.macro.cloud.dao.OrderDao;
import com.macro.cloud.domain.Order;
import com.macro.cloud.service.AccountService;
import com.macro.cloud.service.OrderService;
import com.macro.cloud.service.StorageService;
import io.seata.spring.annotation.GlobalTransactional;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/** * 訂單業務實現類 * Created by macro on 2019/11/11. */
@Service
public class OrderServiceImpl implements OrderService {

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

    @Autowired
    private OrderDao orderDao;
    @Autowired
    private StorageService storageService;
    @Autowired
    private AccountService accountService;

    /** * 建立訂單->調用庫存服務扣減庫存->調用帳戶服務扣減帳戶餘額->修改訂單狀態 */
    @Override
    @GlobalTransactional(name = "fsp-create-order",rollbackFor = Exception.class)
    public void create(Order order) {
        LOGGER.info("------->下單開始");
        //本應用建立訂單
        orderDao.create(order);

        //遠程調用庫存服務扣減庫存
        LOGGER.info("------->order-service中扣減庫存開始");
        storageService.decrease(order.getProductId(),order.getCount());
        LOGGER.info("------->order-service中扣減庫存結束:{}",order.getId());

        //遠程調用帳戶服務扣減餘額
        LOGGER.info("------->order-service中扣減餘額開始");
        accountService.decrease(order.getUserId(),order.getMoney());
        LOGGER.info("------->order-service中扣減餘額結束");

        //修改訂單狀態爲已完成
        LOGGER.info("------->order-service中修改訂單狀態開始");
        orderDao.update(order.getUserId(),0);
        LOGGER.info("------->order-service中修改訂單狀態結束");

        LOGGER.info("------->下單結束");
    }
}
複製代碼

分佈式事務功能演示

  • 運行seata-order-service、seata-storage-service和seata-account-service三個服務；

  • 數據庫初始信息狀態：

  • 咱們在seata-account-service中製造一個超時異常後，調用下單接口：
/** * 帳戶業務實現類 * Created by macro on 2019/11/11. */
@Service
public class AccountServiceImpl implements AccountService {

    private static final Logger LOGGER = LoggerFactory.getLogger(AccountServiceImpl.class);
    @Autowired
    private AccountDao accountDao;

    /** * 扣減帳戶餘額 */
    @Override
    public void decrease(Long userId, BigDecimal money) {
        LOGGER.info("------->account-service中扣減帳戶餘額開始");
        //模擬超時異常，全局事務回滾
        try {
            Thread.sleep(30*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        accountDao.decrease(userId,money);
        LOGGER.info("------->account-service中扣減帳戶餘額結束");
    }
}
複製代碼
  • 此時咱們能夠發現下單後數據庫數據並無任何改變；

  • 咱們能夠在seata-order-service中註釋掉@GlobalTransactional來看看沒有Seata的分佈式事務管理會發生什麼狀況：
/** * 訂單業務實現類 * Created by macro on 2019/11/11. */
@Service
public class OrderServiceImpl implements OrderService {

    /** * 建立訂單->調用庫存服務扣減庫存->調用帳戶服務扣減帳戶餘額->修改訂單狀態 */
    @Override
// @GlobalTransactional(name = "fsp-create-order",rollbackFor = Exception.class)
    public void create(Order order) {
        LOGGER.info("------->下單開始");
        //省略代碼...
        LOGGER.info("------->下單結束");
    }
}
複製代碼
  • 因爲seata-account-service的超時會致使當庫存和帳戶金額扣減後訂單狀態並無設置爲已經完成，並且因爲遠程調用的重試機制，帳戶餘額還會被屢次扣減。

參考資料

Seata官方文檔：github.com/seata/seata…

使用到的模塊

springcloud-learning
├── seata-order-service -- 整合了seata的訂單服務
├── seata-storage-service -- 整合了seata的庫存服務
└── seata-account-service -- 整合了seata的帳戶服務
複製代碼

項目源碼地址

github.com/macrozheng/…

公衆號

mall項目全套學習教程連載中，關注公衆號第一時間獲取。

公衆號圖片
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程