如何在SpringBoot/MySQL事务中并行执行多条SQL？

Question

SpringBoot/MySQL 相关问题
向大佬们请教一个问题，一个事务中要执行多条SQL，比如下单场景，扣减账户余额、修改商品库存、生成流水记录、生成发货记录等操作，假设这些操作比较耗时，有没有什么方法可以让这些SQL在一个事务中并行执行提高性能。

1. 该方案是否类似于分布式事务的解决方案

Answer 1

先写结论：分布式事务在实际应用中并不多见(没有银弹)，一般是采用单机事务 + 异步 + 补偿的方式来解决

首先复杂场景下，应该进一步分析哪些步骤必须保证事务，而哪些步骤不一定要保证事务

如题主说的下单场景中，扣减账户余额和修改商品库存这两个步骤必须保证一致性，所以这两个操作可以放到同一个事务中，而生成流水记录和生成发货记录这两个步骤可能相对耗时且不必须保证事务，所以这两个步骤可以使用异步方式通知下游接口。当然异步调用下游接口时，可能请求接口失败，这时候应有一定的补偿机制，比如通过定时任务来查询已生成的订单且订单状态为未发货的数据，再次请求下游接口进行补偿。

Answer 2

两种方案：
CompletableFuture 异步执行 + TransactionTemplate

@Service
public class OrderService {
    @Autowired 
    private TransactionTemplate transactionTemplate;
    
    public void createOrder() {
        transactionTemplate.execute(status -> {
            try {
                CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
                    // 扣减账户余额
                    accountMapper.reduceBalance();
                });
                
                CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
                    // 修改商品库存  
                    productMapper.updateStock();
                });
                
                CompletableFuture<Void> future3 = CompletableFuture.runAsync(() -> {
                    // 生成流水记录
                    recordMapper.insertRecord(); 
                });
                
                // 等待所有异步任务完成
                CompletableFuture.allOf(future1, future2, future3).join();
                
                return true;
            } catch (Exception e) {
                status.setRollbackOnly();
                throw e;
            }
        });
    }
}

ThreadPoolTaskExecutor + @Transactional

@Service
public class OrderService {
    @Autowired
    private ThreadPoolTaskExecutor executor;
    
    @Transactional
    public void createOrder() {
        List<Future<?>> futures = new ArrayList<>();
        
        futures.add(executor.submit(() -> {
            accountMapper.reduceBalance();
        }));
        
        futures.add(executor.submit(() -> {
            productMapper.updateStock(); 
        }));
        
        futures.add(executor.submit(() -> {
            recordMapper.insertRecord();
        }));
        
        // 等待所有任务完成
        for(Future<?> future : futures) {
            try {
                future.get();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
}

Answer 3

使用异步处理和消息队列来优化性能。

目录结构

spring-boot-order-system/
├── src/
│   ├── main/
│   │   ├── java/
│   │   │   └── com/
│   │   │       └── example/
│   │   │           └── ordersystem/
│   │   │               ├── OrderSystemApplication.java
│   │   │               ├── config/
│   │   │               │   └── RabbitMQConfig.java
│   │   │               ├── controller/
│   │   │               │   └── OrderController.java
│   │   │               ├── entity/
│   │   │               │   ├── Account.java
│   │   │               │   ├── Product.java
│   │   │               │   ├── OrderRecord.java
│   │   │               │   └── ShipmentRecord.java
│   │   │               ├── repository/
│   │   │               │   ├── AccountRepository.java
│   │   │               │   ├── ProductRepository.java
│   │   │               │   ├── OrderRecordRepository.java
│   │   │               │   └── ShipmentRecordRepository.java
│   │   │               ├── service/
│   │   │               │   ├── OrderService.java
│   │   │               │   └── ShipmentService.java
│   │   ├── resources/
│   │   │   ├── application.properties
│   │   │   └── schema.sql
│   └── test/
│       └── java/
│           └── com/
│               └── example/
│                   └── ordersystem/
│                       └── OrderSystemApplicationTests.java
└── pom.xml

1.OrderSystemApplication.java

package com.example.ordersystem;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

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

2.RabbitMQConfig.java

package com.example.ordersystem.config;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    public static final String ORDER_QUEUE = "orderQueue";

    @Bean
    public Queue orderQueue() {
        return new Queue(ORDER_QUEUE, true);
    }
}

3.OrderController.java

package com.example.ordersystem.controller;

import com.example.ordersystem.service.OrderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.math.BigDecimal;

@RestController
@RequestMapping("/orders")
public class OrderController {
    @Autowired
    private OrderService orderService;

    @PostMapping
    public ResponseEntity<String> placeOrder(@RequestParam Long accountId, @RequestParam Long productId, @RequestParam BigDecimal amount) {
        orderService.placeOrder(accountId, productId, amount);
        return ResponseEntity.ok("Order placed successfully");
    }
}

4.实体类
Account.java

package com.example.ordersystem.entity;

import javax.persistence.*;
import java.math.BigDecimal;

@Entity
public class Account {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private BigDecimal balance;
    // getters and setters
}

Product.java

package com.example.ordersystem.entity;

import javax.persistence.*;

@Entity
public class Product {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String name;
    private Integer stock;
    // getters and setters
}

OrderRecord.java

package com.example.ordersystem.entity;

import javax.persistence.*;
import java.math.BigDecimal;
import java.time.LocalDateTime;

@Entity
public class OrderRecord {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private Long accountId;
    private Long productId;
    private BigDecimal amount;
    private LocalDateTime orderTime;
    // getters and setters
}

ShipmentRecord.java

package com.example.ordersystem.entity;

import javax.persistence.*;
import java.time.LocalDateTime;

@Entity
public class ShipmentRecord {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private Long orderId;
    private LocalDateTime shipmentTime;
    // getters and setters
}

5.仓库接口
AccountRepository.java

package com.example.ordersystem.repository;

import com.example.ordersystem.entity.Account;
import org.springframework.data.jpa.repository.JpaRepository;

public interface AccountRepository extends JpaRepository<Account, Long> {
}

ProductRepository.java

package com.example.ordersystem.repository;

import com.example.ordersystem.entity.Product;
import org.springframework.data.jpa.repository.JpaRepository;

public interface ProductRepository extends JpaRepository<Product, Long> {
}

OrderRecordRepository.java

package com.example.ordersystem.repository;

import com.example.ordersystem.entity.OrderRecord;
import org.springframework.data.jpa.repository.JpaRepository;

public interface OrderRecordRepository extends JpaRepository<OrderRecord, Long> {
}

ShipmentRecordRepository.java

package com.example.ordersystem.repository;

import com.example.ordersystem.entity.ShipmentRecord;
import org.springframework.data.jpa.repository.JpaRepository;

public interface ShipmentRecordRepository extends JpaRepository<ShipmentRecord, Long> {
}

6. 服务类
OrderService.java

package com.example.ordersystem.service;

import com.example.ordersystem.entity.*;
import com.example.ordersystem.repository.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.time.LocalDateTime;

@Service
public class OrderService {
    @Autowired
    private AccountRepository accountRepository;
    @Autowired
    private ProductRepository productRepository;
    @Autowired
    private OrderRecordRepository orderRecordRepository;
    @Autowired
    private ShipmentRecordRepository shipmentRecordRepository;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Transactional
    public void placeOrder(Long accountId, Long productId, BigDecimal amount) {
        // 扣减账户余额
        Account account = accountRepository.findById(accountId).orElseThrow();
        account.setBalance(account.getBalance().subtract(amount));
        accountRepository.save(account);

        // 修改商品库存
        Product product = productRepository.findById(productId).orElseThrow();
        product.setStock(product.getStock() - 1);
        productRepository.save(product);

        // 生成订单流水记录
        OrderRecord orderRecord = new OrderRecord();
        orderRecord.setAccountId(accountId);
        orderRecord.setProductId(productId);
        orderRecord.setAmount(amount);
        orderRecord.setOrderTime(LocalDateTime.now());
        orderRecordRepository.save(orderRecord);

        // 发送发货任务到消息队列
        rabbitTemplate.convertAndSend(RabbitMQConfig.ORDER_QUEUE, orderRecord.getId());
    }
}

ShipmentService.java

package com.example.ordersystem.service;

import com.example.ordersystem.entity.ShipmentRecord;
import com.example.ordersystem.repository.OrderRecordRepository;
import com.example.ordersystem.repository.ShipmentRecordRepository;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;

@Service
public class ShipmentService {
    @Autowired
    private ShipmentRecordRepository shipmentRecordRepository;
    @Autowired
    private OrderRecordRepository orderRecordRepository;

    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void handleShipment(Long orderId) {
        // 生成发货记录
        ShipmentRecord shipmentRecord = new ShipmentRecord();
        shipmentRecord.setOrderId(orderId);
        shipmentRecord.setShipmentTime(LocalDateTime.now());
        shipmentRecordRepository.save(shipmentRecord);
    }
}

7. 配置文件
application.properties

spring.datasource.url=jdbc:mysql://localhost:3306/order_system
spring.datasource.username=root
spring.datasource.password=root
spring.jpa.hibernate.ddl-auto=update

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

schema.sql

CREATE TABLE account (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    balance DECIMAL(19, 2)
);

CREATE TABLE product (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255),
    stock INT
);

CREATE TABLE order_record (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    account_id BIGINT,
    product_id BIGINT,
    amount DECIMAL(19, 2),
    order_time TIMESTAMP
);

CREATE TABLE shipment_record (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    order_id BIGINT,
    shipment_time TIMESTAMP
);

8. 测试类
OrderSystemApplicationTests.java

package com.example.ordersystem;

import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class OrderSystemApplicationTests {

    @Test
    void contextLoads() {
    }
}

9. 项目构建文件
pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>order-system</artifactId>
    <version>1.0.0</version>
    <packaging>jar</packaging>

    <name>order-system</name>
    <description>Spring Boot Order System</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.4</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <java.version>11</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-data-jpa</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

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