初识Seata（二）

目录

1.背景

2.环境

3.业务模型

4.数据库准备

1）订单库

2）库存库

3）账户库

5.微服务准备

1）pom.xml

2）注册中心准备

3）订单服务

1.背景

书接上回：初识Seata（一），本文以一个Seata AT模式为例，介绍Seata的使用。

2.环境

SpringBoot	2.1.2.RELEASE
SpringCloud	Greenwich.SR1
spring-cloud-alibaba-dependencies	2.1.0.RELEASE
seata	1.2.0
mysql	5.1.47

3.业务模型

下订单 --> 扣减库存 --> 扣减账户余额 --> 修改订单状态

4.数据库准备

1）订单库

CREATE DATAbase IF NOT EXISTS `seata_order`
USE `seata_order`;

CREATE DATAbase IF NOT EXISTS `orders` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `user_id` bigint(20) DEFAULT NULL,
  `product_id` bigint(20) DEFAULT NULL,
  `count` int(11) DEFAULT NULL COMMENT '数量',
  `money` decimal(10,2) DEFAULT NULL,
  `status` varchar(100) DEFAULT NULL,
  `gmt_create` datetime DEFAULT CURRENT_TIMESTAMP,
  `gmt_modified` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=55 DEFAULT CHARSET=utf8;


CREATE TABLE IF NOT EXISTS `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`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

2）库存库

CREATE DATAbase IF NOT EXISTS `seata_storage`
USE `seata_storage`;

CREATE DATAbase IF NOT EXISTS `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;


CREATE TABLE IF NOT EXISTS `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`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

3）账户库

CREATE DATAbase IF NOT EXISTS `seata_account`
USE `seata_account`;

CREATE DATAbase IF NOT EXISTS `account` (
  `id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT 'id',
  `user_id` bigint(11) DEFAULT NULL COMMENT '用户id',
  `total` decimal(10,2) DEFAULT NULL COMMENT '总额度',
  `used` decimal(10,2) DEFAULT NULL COMMENT '已用额度',
  `balance` decimal(10,2) DEFAULT '0.00' COMMENT '剩余可用额度',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;


CREATE TABLE IF NOT EXISTS `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`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

5.微服务准备

1）pom.xml

基础的包如下，不再赘述：spring-boot-dependencies/ spring-cloud-dependencies/ spring-cloud-alibaba-dependencies/ mysql-connector-java/ mybatis-spring-boot-starter/ druid/ druid-spring-boot-starter/ junit/ log4j/ lombok/ javax.annotation-api

关于seata的依赖包：

    com.alibaba.cloud
    spring-cloud-alibaba-seata
    2.1.0.RELEASE
    
        
            seata-all
            io.seata
        
    


    io.seata
    seata-all
    1.2.0

2）注册中心准备

*** pom.xml就不再赘述了

application.yml

server:
  port: 8801

eureka:
  instance:
    hostname: localhost
    prefer-ip-address: true
  client:
    service-url:
      defaultZone: http://localhost:8801/eureka/
    registerWithEureka: false
    fetchRegistry: false

 启动类

@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
@EnableEurekaServer
public class EurekaAppMain8801
{
    public static void main(String[] args){
        SpringApplication.run(EurekaAppMain8801.class, args);
    }
}

3）订单服务

目录结构

java

com.seata.order

configcontrollerdaoentityserviceresources

mapper（package）application.ymlfile.confregistry.conf

*** file.conf

transport {
    type = "TCP"
    server = "NIO"
    heartbeat = true
    enableClientBatchSendRequest = true
    threadFactory {
        bossThreadPrefix = "NettyBoss"
        workerThreadPrefix = "NettyServerNIOWorker"
        serverExecutorThread-prefix = "NettyServerBizHandler"
        shareBossWorker = false
        clientSelectorThreadPrefix = "NettyClientSelector"
        clientSelectorThreadSize = 1
        clientWorkderThreadPrefix = "NettyClientWorkerThread"
        bossThreadSize = 1
        workerThreadSize = "default"
    }
    shutdown {
        wait = 3
    }
    serialzation = "seata"
    compressor = "none"
}
service {
    vgroupMapping.my_test_tx_group = "seata-server"  #  这里的my_test_tx_group与application.yml中保持一致
    seata-server.grouplist = "127.0.0.1:8091"
    enableDegrade = false
    disableGlobalTransaction = false
}
client {
  rm {
    asyncCommitBufferLimit = 10000
    lock {
        retryInterval = 10
        retryTimes = 30
        retryPolicyBranchRollbackConflict = true
    }
    reportRetryCount = 5
    tablemetaCheckEnable = false
    reportSuccessEnable = false
  }
  tm {
    commitRetryCount = 5
    rollbackRetryCount = 5
  }
  undo {
    dataValidation = true
    logSerialization = "jackson"
    logTable = "undo_log"
  }
  log {
    exceptionRate = 100
  }
}

*** registry.conf

registry {
  # 可选用的类型file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  type = "eureka"
 
  nacos {
    application = "seata-server"
    serverAddr = "localhost"
    namespace = ""
    cluster = "default"
    username = ""
    password = ""
  }
  eureka {
    serviceUrl = "http://localhost:8801/eureka"
    application = "seata-server"
    weight = "1"
  }
  redis {
    serverAddr = "localhost:6379"
    db = 0
    password = ""
    cluster = "default"
    timeout = 0
  }
  zk {
    cluster = "default"
    serverAddr = "127.0.0.1:2181"
    sessionTimeout = 6000
    connectTimeout = 2000
    username = ""
    password = ""
  }
  consul {
    cluster = "default"
    serverAddr = "127.0.0.1:8500"
  }
  etcd3 {
    cluster = "default"
    serverAddr = "http://localhost:2379"
  }
  sofa {
    serverAddr = "127.0.0.1:9603"
    application = "default"
    region = "DEFAULT_ZONE"
    datacenter = "DefaultDataCenter"
    cluster = "default"
    group = "SEATA_GROUP"
    addressWaitTime = "3000"
  }
  file {
    name = "file.conf"
  }
}
config {
  # file、nacos 、apollo、zk、consul、etcd3、springCloudConfig
  type = "file"
 
  nacos {
    serverAddr = "localhost"
    namespace = ""
    group = "SEATA_GROUP"
    username = ""
    password = ""
  }
  consul {
    serverAddr = "127.0.0.1:8500"
  }
  apollo {
    appId = "seata-server"
    apollometa = "http://192.168.1.204:8801"
    namespace = "application"
  }
  zk {
    serverAddr = "127.0.0.1:2181"
    sessionTimeout = 6000
    connectTimeout = 2000
    username = ""
    password = ""
  }
  etcd3 {
    serverAddr = "http://localhost:2379"
  }
  file {
    name = "file.conf"
  }
}

*** application.yml

server:
  port: 8802
spring:
  application:
    name: seata-order
  cloud:
    alibaba:
     seata:
       # 这里要与file.conf中的值保持一致
       tx-service-group: my_test_tx_group
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    username: root
    password: 123456
    url: jdbc:mysql://locahost:3306/seata_order?serverTimezone=GMT%2B8&characterEncoding=utf8

eureka:
  client:
    service-url:
      dafaultZone: http://localhost:8801/eureka/
  instance:
    hostname: localhost
    prefer-ip-address: true
logging:
  level:
    io:
      seata: info
mybatis:
  mapperLocations: classpath:mapper/*.xml

*** 启动类

@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
@EnableDiscoveryClient
@EnableEurekaClients
public class OrderAppMain8802
{
    public static void main(String[] args){
        SpringApplication.run(OrderAppMain8802.class, args);
    }
}

*** config

**** DataSourceProxyConfig.java

@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 sqlSessionFactory(DataSourceProxy  dataSourceProxy) throws Exception {
        SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
        sqlSessionFactoryBean.setDataSource(dataSourceProxy);
        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources(mapperLocations));
        sqlSessionFactoryBean.setTransactionFactory(new SpringManagedTransactionFactory());
        return sqlSessionFactoryBean.getObject();
    }
}

**** MyBatisConfig.java

@Configuration
@MapperScan({"com.seata.order.dao"})
public class MyBatisConfig{
}

*** entity

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Order {
    private Long id;
    private Long userId;
    private Long productId;
    private Integer count;
    private BigDecimal money;
    private Integer status;
    private Date gmtCreate;
    private Date gmtModified;
}

*** controller

@RequestController
@RequestMapping("/order/")
public class OrderController {
    @Resource
    private OrderService orderService;

    @GetMapping("create")
    public String create(Order order){
        orderService.create(order);
        return "订单创建成功，库存扣减成功，账户余额扣减成功"
    }
}

*** service & serviceImpl

**** service

public interface OrderService {
    void create(Order order);
}

@FeignClient(name = "account-service")
public interface AccountService {
    @GetMapping(value = "/account/decrease")
    String decrease(@RequestParam("userId") Long userId, @RequestParam("money") BigDecimal money);
}

@FeignClient(name = "storage-service")
public interface StorageService {
    @GetMapping(value = "/storage/decrease")
    String decrease(@RequestParam("productId") Long productId, @RequestParam("count") Integer count);
}

**** serviceImpl

@Service
public class OrderServiceImpl implements OrderService {
    @Resource
    private OrderDao orderDao;

    @Resource
    private StorageService storageService;

    @Resource
    private AccountService accountService;

    // 这个注解很重要
    @GlobalTransactional(name = "app-create-order", rollback = Exception.class)
    public void create(Order order) {
        orderDao.create(order);
        storageService.decrease(order.getProductId(), order.getCount());
        accountService.decrease(order.getUserId(), order.getMoney());
        orderDao.update(order.getUserId(), order.getMoney(), 0);
    }
}

*** dao

@Mapper
public interface OrderDao {
    void create(Order order);
}

*** resources

**** mapper.OrderMapper.xml

 

    
        
        
        
        
        
        
        
        
    
    
        insert into orders(user_id, product_id, count, money, status)
        values (#{userId}, #{productId}, #{count}, #{money}, 0)
    

未完待续~~