分布式事务介绍
在分布式系统中实现的事务就是分布式事务,分布式系统的CAP原则是:
是分布式事务主要是保证数据的一致性,主要有三种不同的原则
共同点:
Orderservice监听新订单队列中的消息,获取之后新增订单,成功则往新订单缴费队列中写消息,中间新增订单的过程使用JTA事务管理,当新增失败则事务回滚,不会往新订单缴费队列中写消息;
再比如User service 扣费成功后,往新订单转移票队列写消息,这时Ticket service 正在处理中或者处理中发生了失败,这中间的过程中用户查看自己的余额已经扣费成功,但票的信息却没有,此时可以使用事务失败回滚的方式依次回退,这种叫弱一致性;又或者可以把处理失败的内容发送至一个错误队列中,由人工处理等方式解决,这种叫最终一致性。
因为JTA采用两阶段提交方式,第一次是预备阶段,第二次才是正式提交。当第一次提交出现错误,则整个事务出现回滚,一个事务的时间可能会较长,因为它要跨越多个数据库多个数据资源的的操作,所以在性能上可能会造成吞吐量低。
1.start message transaction
2.receive message
3.start database transaction
4.update database
5.commit database transaction
6.commit message transaction ##当这一步出现错误时,上面的因为已经commit,所以不会rollback
这时候就会出现问题
1.start message transaction
2.receive message
3.start JTA transaction on DB
4.update database
5.phase-1 commit on DB transaction
6.commit message transaction ##当这一步出现错误时,上面的因为是XA的第一次提交预备状态,所以可以rollback
7.phase-2 commit on DB transaction ##当这一步出现错误时,因为message不是XA方式,commit后无法rollback
但这种相比不使用JTA,已经很大程度上避免了事务发生错误的可能性。
1.start message transaction
2.receive message
3.start database transaction
4.update database #数据库操作出错,消息被放回MQ队列,重试重新触发该方法
5.commit database transaction
6.commit message transaction
上面这种时候没有问题
1.start message transaction
2.receive message
3.start database transaction
4.update database
5.commit database transaction
6.commit message transaction #提交MQ事务出错,消息放回至MQ队列,重试重新触发该方法
可能存在问题:会重复数据库操作,因为database transaction不是使用JTA事务管理,所以database已经commit成功;如何避免,需要忽略重发消息,比如唯一性校验等手段。
application.properties中配置了两个数据源
# 默认的Datasource配置
# spring.datasource.url = jdbc:mysql://localhost:3307/user
# spring.datasource.username = root
# spring.datasource.password = 123456
# spring.datasource.driverClassName = com.mysql.jdbc.Driver
spring.ds_user.url = jdbc:mysql://localhost:3307/js_user
spring.ds_user.username = root
spring.ds_user.password = 123456
spring.ds_user.driver-class-name = com.mysql.jdbc.Driver
spring.ds_order.url = jdbc:mysql://localhost:3307/js_order
spring.ds_order.username = root
spring.ds_order.password = 123456
spring.ds_order.driver-class-name = com.mysql.jdbc.Driver
自定义配置类文件
@Configuration
public class DBConfiguration{
@Bean
@Primary
@ConfigurationProperties(prefix="spring.ds_user") #设置读取在properties文件内容的前缀
public DataSourceProperties userDataSourceProperties() {
return new DataSourceProperties();
}
@Bean
@Primary
public DataSource userDataSource(){
return userDataSourceProperties().initializeDataSourceBuilder().type(HikariDataSource.class).build();
}
@Bean
public JdbcTemplate userJdbcTemplate(@Qualifier("userDataSource") DataSource userDataSource){
return new JdbcTemplate(userDataSource);
}
@Bean
@ConfigurationProperties(prefix="spring.ds_order") #设置读取在properties文件内容的前缀
public DataSourceProperties orderDataSourceProperties() {
return new DataSourceProperties();
}
@Bean
public DataSource orderDataSource(){
return userDataSourceProperties().initializeDataSourceBuilder().type(HikariDAtaSource.class).build();
}
@Bean
public JdbcTemplate orderJdbcTemplate(@Qualifier("orderDataSource") DataSource orderDataSource){
return new JdbcTemplate(orderDataSource);
}
}
复制代码
实际调用类
public class CustomerService{
@Autowired
@Qualifier("userJdbcTemplate")
private jdbcTemplate userJdbcTemplate;
@Autowired
@Qualifier("orderJdbcTemplate")
private jdbcTemplate orderJdbcTemplate;
private static final String UPDATE_CUSTOMER_SQL;
private static final String INSERT_ORDER_SQL;
@Transactional #事务管理注解
public void createOrder(Order order){
userJdbcTemplate.update(UPDATE_CUSTOMER_SQL, order)
if(order.getTitle().contains("error1")){ #模拟异常出现
throw new RuntimeException("error1")
}
orderJdbcTemplate.update(INSERT_ORDER_SQL, order) #没有使用事务,直接提交
if(order.getTitle().contains("error2")){ #模拟异常出现
throw new RuntimeException("error2")
}
}
}
关于上述过程的详细说明:
因为使用了标签 @Transactional的方式,使其在一个事务里面执行
也就是同步到Transaction Manager上面,但是这边的同步不是说事务的同步,只是同步数据库连接的开关
特别说明: @Transactional 如果没有做任何配置的情况下,则会使用DBConfiguration类中@Primart注解下的DataSource,用它去做datasource connection
spring DataSourceUtils 使用已有的connection,只是控制数据库连接的释放,不是事务。
链式事务管理器在 这个库里面
DBConfiguration类中添加一段
@Bean
public PlatformTransactionManager transactionManager(){
DataSourceTransactionManager userTM = new DataSourceTransactionManager(userDataSource()) #看似方法调用,实则从spring容器中获取
DataSourceTransactionManager orderTM = new DataSourceTransactionManager(orderDataSource())
# orderTM.setDataSource(orderDataSource()) 如果使用这种方式则不是从容器中去获取了,因为orderTM不是spring容器管理
ChainedTransactionManager tm = new ChainedTransactionManager(userTM, orderTM) ## order先执行,user后执行
return tm;
}
复制代码
链接事务管理器(Chaining transaction managers)
出现异常是否会有问题呢?
重启启动msyql服务,程序继续运行,此时来看数据库order表中多了一条记录,而user表没有变化;第一个order事务并没有回滚;那如果是rollback的时候停掉mysql服务,其实是没有影响的,因为本身就没有commit, 执不执行rollback本身是没有影响的。
git代码地址 ☚
git代码地址 ☚
public OrderService{
Map disMap; # 用于存放已经处理的id
@Transactional
void ticketOrder(BuyTickerDTO dto){
String uid = createUUID(dto); # 创建并获取数据的唯一id
if(!diMap.contains(uuid){ #disMap还没有处理过这个数据唯一id,则进入创建
Order order = createOrder(dto);
disMap.append(uid) ## 追加Map
}
}
userService.charge(dto); #调用user微服务
}
#通过调节限定,只有第一次支付的时候才会扣余额,被重复调用的时候就不会重复扣费用,通过paystatus判断
UPDATE customer SET deposit = deposit - ${value}, paystatus = 'PAID' WHERE orderId = ${id} and paystatus = 'UNPAID'