- 微信
- 微博
  
  分享文章到微博
- 复制链接
  
  复制链接到剪贴板

Java 异步并发处理数据集事务问题排查实战

bug菌发表于 2024/12/31 10:04:04 2024/12/31

【摘要】 🏆本文收录于「滚雪球学SpringBoot」专栏，专业攻坚指数级提升持续更新中，up！up！up！！欢迎点赞&&收藏&&订阅。@TOC ✨ 前言在现代业务系统中，异步并发处理已成为高效处理海量数据的关键手段。然而，当涉及数据库事务（Transaction）时，异步并发处理会引发许多挑战和问题，比如脏读、幻读、死锁等。本文通过一个实战案例，深入分析如何在 Java 异步并发场景下排查和解决事...

🏆本文收录于「滚雪球学SpringBoot」专栏，专业攻坚指数级提升持续更新中，up！up！up！！欢迎点赞&&收藏&&订阅。

@TOC

✨ 前言

在现代业务系统中，异步并发处理已成为高效处理海量数据的关键手段。然而，当涉及数据库事务（Transaction）时，异步并发处理会引发许多挑战和问题，比如脏读、幻读、死锁等。

本文通过一个实战案例，深入分析如何在 Java 异步并发场景下排查和解决事务相关问题。

🧠 常见的事务问题

在异步并发处理数据时，可能遇到以下问题：

脏读：
- 一个事务读取到了另一个未提交事务的数据。
幻读：
- 一个事务两次查询的结果不一致，因为另一事务插入了新的数据。
不可重复读：
- 一个事务两次读取同一数据，因另一个事务更新导致结果不一致。
死锁：
- 两个或多个事务相互等待对方持有的资源而无法继续。
事务失效：
- 异步任务没有正确传播事务上下文。

🌟 实战案例：并发处理订单的事务问题

场景描述：

一个电商系统需要批量处理订单状态：

从数据库中查询所有未支付的订单。
对每个订单进行异步并发处理，将状态从 PENDING 修改为 CANCELLED。
要求事务一致性，确保所有订单的状态正确更新。

问题：

在测试中，发现部分订单的状态未正确更新，甚至有状态被误改的情况。

⚙️ 问题分析与排查

1. 使用的数据库事务隔离级别

数据库事务的隔离级别会直接影响事务的行为。常见的隔离级别有：

READ UNCOMMITTED（读取未提交数据，容易引发脏读）
READ COMMITTED（避免脏读，但无法避免不可重复读）
REPEATABLE READ（避免脏读和不可重复读，但可能出现幻读）
SERIALIZABLE（完全隔离，性能较低）

排查步骤：

确认当前数据库的默认隔离级别，检查是否适合业务场景：
```
SELECT @@transaction_isolation;
```
如果隔离级别过低（如 READ UNCOMMITTED），应调整为 READ COMMITTED 或更高。

2. 是否正确使用事务传播机制

Spring 中事务的传播机制（Propagation）是事务管理的核心。常见模式包括：

REQUIRED：加入当前事务，若无事务则创建新事务。
REQUIRES_NEW：挂起当前事务，创建新事务。
SUPPORTS：支持当前事务，若无事务则以非事务方式执行。

问题示例：

在异步方法中未正确传播事务：

@Transactional
public void processOrdersAsync(List<Order> orders) {
    orders.forEach(order -> CompletableFuture.runAsync(() -> updateOrder(order)));
}

private void updateOrder(Order order) {
    // 此方法未被事务管理，可能引发事务问题
    order.setStatus("CANCELLED");
    orderRepository.save(order);
}

3. 是否存在并发更新冲突

多线程环境中，多个线程可能尝试同时更新同一条记录，引发写冲突或死锁问题。

问题示例：

多个线程同时更新订单：

@Modifying
@Query("UPDATE Order o SET o.status = 'CANCELLED' WHERE o.id = :orderId AND o.status = 'PENDING'")
void cancelOrder(@Param("orderId") Long orderId);

如果两个线程同时执行该 SQL，可能发生以下情况：

线程 A：读取状态为 PENDING，并尝试更新。
线程 B：在线程 A 提交前也读取到了 PENDING 状态，尝试更新相同记录。

解决方案是引入乐观锁或悲观锁。

4. 异步任务未捕获异常

异步任务中如果未正确捕获异常，可能导致部分任务失败，事务回滚不完整。

问题示例：

异步任务中未捕获异常：

CompletableFuture.runAsync(() -> {
    order.setStatus("CANCELLED");
    orderRepository.save(order);
});

如果 save 操作抛出异常，该任务将直接失败，且不会影响其他任务。

🛠️ 解决方案

1. 使用正确的事务隔离级别

针对高并发环境，推荐使用 REPEATABLE READ 或更高的 SERIALIZABLE 隔离级别：

@Transactional(isolation = Isolation.REPEATABLE_READ)
public void processOrders() {
    List<Order> orders = orderRepository.findPendingOrders();
    orders.forEach(order -> {
        order.setStatus("CANCELLED");
        orderRepository.save(order);
    });
}

2. 使用事务传播机制

确保异步任务在事务上下文中执行。可以通过以下方式实现：

方式 1：显式事务管理

使用 @Transactional 注解，并在异步任务中手动调用事务上下文：

@Transactional
public void processOrders(List<Order> orders) {
    orders.forEach(order -> {
        CompletableFuture.runAsync(() -> updateOrderWithTransaction(order));
    });
}

@Transactional
public void updateOrderWithTransaction(Order order) {
    order.setStatus("CANCELLED");
    orderRepository.save(order);
}

方式 2：结合 Spring 的 `@Async`

将异步任务交给 Spring 的 @Async 注解管理：

@Service
public class AsyncOrderService {

    @Async
    @Transactional
    public void updateOrderAsync(Order order) {
        order.setStatus("CANCELLED");
        orderRepository.save(order);
    }
}

调用：

orders.forEach(order -> asyncOrderService.updateOrderAsync(order));

3. 使用乐观锁避免并发冲突

步骤：

在订单表中添加版本号字段（version）。

ALTER TABLE orders ADD COLUMN version INT DEFAULT 0;

修改实体类，添加 @Version 注解：

@Entity
public class Order {
    @Id
    private Long id;

    @Version
    private Integer version;

    private String status;

    // Getters and Setters
}

更新方法中，启用乐观锁：

public void updateOrder(Order order) {
    order.setStatus("CANCELLED");
    orderRepository.save(order);
}

当多个线程尝试更新同一记录时，@Version 注解会检测版本冲突，抛出 OptimisticLockException，从而避免数据不一致。

4. 捕获和处理异步异常

为每个异步任务设置异常处理逻辑：

orders.forEach(order -> CompletableFuture.runAsync(() -> {
    try {
        order.setStatus("CANCELLED");
        orderRepository.save(order);
    } catch (Exception e) {
        System.err.println("Failed to process order: " + order.getId());
        e.printStackTrace();
    }
}));

或使用 handle 方法处理异常：

CompletableFuture.runAsync(() -> {
    order.setStatus("CANCELLED");
    orderRepository.save(order);
}).handle((result, ex) -> {
    if (ex != null) {
        System.err.println("Error: " + ex.getMessage());
    }
    return null;
});

✨ 完整解决方案

以下是结合上述方法的完整代码：

@Transactional
public void processOrders() {
    List<Order> orders = orderRepository.findPendingOrders();
    orders.forEach(order -> {
        CompletableFuture.runAsync(() -> {
            try {
                updateOrderWithTransaction(order);
            } catch (Exception e) {
                System.err.println("Failed to process order: " + order.getId());
                e.printStackTrace();
            }
        });
    });
}

@Transactional
public void updateOrderWithTransaction(Order order) {
    order.setStatus("CANCELLED");
    orderRepository.save(order);
}

✨ 总结

在 Java 异步并发处理数据时，事务问题的排查和解决需要从以下几个方面入手：

选择合适的事务隔离级别，避免脏读、幻读和不可重复读问题。
正确传播事务上下文，确保异步任务受到事务管理。
引入乐观锁或悲观锁，解决并发更新冲突。
捕获异步任务中的异常，避免任务失败后未被感知。

通过合理设计和优化，可以有效提高系统的并发处理能力，同时保证数据一致性和完整性。🎉

✨ 前言

本文通过一个实战案例，深入分析如何在 Java 异步并发场景下排查和解决事务相关问题。

🧠 常见的事务问题

在异步并发处理数据时，可能遇到以下问题：

脏读：
- 一个事务读取到了另一个未提交事务的数据。
幻读：
- 一个事务两次查询的结果不一致，因为另一事务插入了新的数据。
不可重复读：
- 一个事务两次读取同一数据，因另一个事务更新导致结果不一致。
死锁：
- 两个或多个事务相互等待对方持有的资源而无法继续。
事务失效：
- 异步任务没有正确传播事务上下文。

🌟 实战案例：并发处理订单的事务问题

场景描述：

一个电商系统需要批量处理订单状态：

从数据库中查询所有未支付的订单。
对每个订单进行异步并发处理，将状态从 PENDING 修改为 CANCELLED。
要求事务一致性，确保所有订单的状态正确更新。

问题：

在测试中，发现部分订单的状态未正确更新，甚至有状态被误改的情况。

⚙️ 问题分析与排查

1. 使用的数据库事务隔离级别

数据库事务的隔离级别会直接影响事务的行为。常见的隔离级别有：

READ UNCOMMITTED（读取未提交数据，容易引发脏读）
READ COMMITTED（避免脏读，但无法避免不可重复读）
REPEATABLE READ（避免脏读和不可重复读，但可能出现幻读）
SERIALIZABLE（完全隔离，性能较低）

排查步骤：

确认当前数据库的默认隔离级别，检查是否适合业务场景：
```
SELECT @@transaction_isolation;
```
如果隔离级别过低（如 READ UNCOMMITTED），应调整为 READ COMMITTED 或更高。

2. 是否正确使用事务传播机制

Spring 中事务的传播机制（Propagation）是事务管理的核心。常见模式包括：

REQUIRED：加入当前事务，若无事务则创建新事务。
REQUIRES_NEW：挂起当前事务，创建新事务。
SUPPORTS：支持当前事务，若无事务则以非事务方式执行。

问题示例：

在异步方法中未正确传播事务：

@Transactional
public void processOrdersAsync(List<Order> orders) {
    orders.forEach(order -> CompletableFuture.runAsync(() -> updateOrder(order)));
}

private void updateOrder(Order order) {
    // 此方法未被事务管理，可能引发事务问题
    order.setStatus("CANCELLED");
    orderRepository.save(order);
}

3. 是否存在并发更新冲突

多线程环境中，多个线程可能尝试同时更新同一条记录，引发写冲突或死锁问题。

问题示例：

多个线程同时更新订单：

@Modifying
@Query("UPDATE Order o SET o.status = 'CANCELLED' WHERE o.id = :orderId AND o.status = 'PENDING'")
void cancelOrder(@Param("orderId") Long orderId);

如果两个线程同时执行该 SQL，可能发生以下情况：

线程 A：读取状态为 PENDING，并尝试更新。
线程 B：在线程 A 提交前也读取到了 PENDING 状态，尝试更新相同记录。

解决方案是引入乐观锁或悲观锁。

4. 异步任务未捕获异常

异步任务中如果未正确捕获异常，可能导致部分任务失败，事务回滚不完整。

问题示例：

异步任务中未捕获异常：

CompletableFuture.runAsync(() -> {
    order.setStatus("CANCELLED");
    orderRepository.save(order);
});

如果 save 操作抛出异常，该任务将直接失败，且不会影响其他任务。

🛠️ 解决方案

1. 使用正确的事务隔离级别

针对高并发环境，推荐使用 REPEATABLE READ 或更高的 SERIALIZABLE 隔离级别：

@Transactional(isolation = Isolation.REPEATABLE_READ)
public void processOrders() {
    List<Order> orders = orderRepository.findPendingOrders();
    orders.forEach(order -> {
        order.setStatus("CANCELLED");
        orderRepository.save(order);
    });
}

2. 使用事务传播机制

确保异步任务在事务上下文中执行。可以通过以下方式实现：

方式 1：显式事务管理

使用 @Transactional 注解，并在异步任务中手动调用事务上下文：

@Transactional
public void processOrders(List<Order> orders) {
    orders.forEach(order -> {
        CompletableFuture.runAsync(() -> updateOrderWithTransaction(order));
    });
}

@Transactional
public void updateOrderWithTransaction(Order order) {
    order.setStatus("CANCELLED");
    orderRepository.save(order);
}

方式 2：结合 Spring 的 `@Async`

将异步任务交给 Spring 的 @Async 注解管理：

@Service
public class AsyncOrderService {

    @Async
    @Transactional
    public void updateOrderAsync(Order order) {
        order.setStatus("CANCELLED");
        orderRepository.save(order);
    }
}

调用：

orders.forEach(order -> asyncOrderService.updateOrderAsync(order));

3. 使用乐观锁避免并发冲突

步骤：

在订单表中添加版本号字段（version）。

ALTER TABLE orders ADD COLUMN version INT DEFAULT 0;

修改实体类，添加 @Version 注解：

@Entity
public class Order {
    @Id
    private Long id;

    @Version
    private Integer version;

    private String status;

    // Getters and Setters
}

更新方法中，启用乐观锁：

public void updateOrder(Order order) {
    order.setStatus("CANCELLED");
    orderRepository.save(order);
}

当多个线程尝试更新同一记录时，@Version 注解会检测版本冲突，抛出 OptimisticLockException，从而避免数据不一致。

4. 捕获和处理异步异常

为每个异步任务设置异常处理逻辑：

orders.forEach(order -> CompletableFuture.runAsync(() -> {
    try {
        order.setStatus("CANCELLED");
        orderRepository.save(order);
    } catch (Exception e) {
        System.err.println("Failed to process order: " + order.getId());
        e.printStackTrace();
    }
}));

或使用 handle 方法处理异常：

CompletableFuture.runAsync(() -> {
    order.setStatus("CANCELLED");
    orderRepository.save(order);
}).handle((result, ex) -> {
    if (ex != null) {
        System.err.println("Error: " + ex.getMessage());
    }
    return null;
});

✨ 完整解决方案

以下是结合上述方法的完整代码：

@Transactional
public void processOrders() {
    List<Order> orders = orderRepository.findPendingOrders();
    orders.forEach(order -> {
        CompletableFuture.runAsync(() -> {
            try {
                updateOrderWithTransaction(order);
            } catch (Exception e) {
                System.err.println("Failed to process order: " + order.getId());
                e.printStackTrace();
            }
        });
    });
}

@Transactional
public void updateOrderWithTransaction(Order order) {
    order.setStatus("CANCELLED");
    orderRepository.save(order);
}

✨ 总结

在 Java 异步并发处理数据时，事务问题的排查和解决需要从以下几个方面入手：

选择合适的事务隔离级别，避免脏读、幻读和不可重复读问题。
正确传播事务上下文，确保异步任务受到事务管理。
引入乐观锁或悲观锁，解决并发更新冲突。
捕获异步任务中的异常，避免任务失败后未被感知。

通过合理设计和优化，可以有效提高系统的并发处理能力，同时保证数据一致性和完整性。🎉

🧧福利赠与你🧧

无论你是计算机专业的学生，还是对编程有兴趣的小伙伴，都建议直接毫无顾忌的学习此专栏「滚雪球学SpringBoot」，bug菌郑重承诺，凡是学习此专栏的同学，均能获取到所需的知识和技能，全网最快速入门SpringBoot，就像滚雪球一样，越滚越大，无边无际，指数级提升。

最后，如果这篇文章对你有所帮助，帮忙给作者来个一键三连，关注、点赞、收藏，您的支持就是我坚持写作最大的动力。

同时欢迎大家关注公众号:「猿圈奇妙屋」，以便学习更多同类型的技术文章，免费白嫖最新BAT互联网公司面试题、4000G pdf电子书籍、简历模板、技术文章Markdown文档等海量资料。

✨️ Who am I?

我是bug菌，CSDN | 掘金 | InfoQ | 51CTO | 华为云 | 阿里云 | 腾讯云等社区博客专家，C站博客之星Top30，华为云2023年度十佳博主，掘金多年度人气作者Top40，掘金等各大社区平台签约作者，51CTO年度博主Top12，掘金/InfoQ/51CTO等社区优质创作者；全网粉丝合计 30w+；更多精彩福利点击这里；硬核微信公众号「猿圈奇妙屋」，欢迎你的加入！免费白嫖最新BAT互联网公司面试真题、4000G PDF电子书籍、简历模板等海量资料，你想要的我都有，关键是你不来拿。

-End-

【声明】本内容来自华为云开发者社区博主，不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息，否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com

点赞
收藏
关注作者

0/1000

抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认取消

加入云驻计划，成为创作者

华为云周边好礼
免费体验产品
特殊身份标识
线下官方门票
内部专家零距离
与10000+优质创作者共同成长

立即加入

Java 异步并发处理数据集事务问题排查实战

✨ 前言

🧠 常见的事务问题

🌟 实战案例：并发处理订单的事务问题

场景描述：

问题：

⚙️ 问题分析与排查

1. 使用的数据库事务隔离级别

排查步骤：

2. 是否正确使用事务传播机制

问题示例：

3. 是否存在并发更新冲突

问题示例：

4. 异步任务未捕获异常

问题示例：

🛠️ 解决方案

1. 使用正确的事务隔离级别

2. 使用事务传播机制

方式 1：显式事务管理

方式 2：结合 Spring 的 @Async

3. 使用乐观锁避免并发冲突

步骤：

4. 捕获和处理异步异常

✨ 完整解决方案

✨ 总结

✨ 前言

🧠 常见的事务问题

🌟 实战案例：并发处理订单的事务问题

场景描述：

问题：

⚙️ 问题分析与排查

1. 使用的数据库事务隔离级别

排查步骤：

2. 是否正确使用事务传播机制

问题示例：

3. 是否存在并发更新冲突

问题示例：

4. 异步任务未捕获异常

问题示例：

🛠️ 解决方案

1. 使用正确的事务隔离级别

2. 使用事务传播机制

方式 1：显式事务管理

方式 2：结合 Spring 的 @Async

3. 使用乐观锁避免并发冲突

步骤：

4. 捕获和处理异步异常

✨ 完整解决方案

✨ 总结

🧧福利赠与你🧧

✨️ Who am I?

全部回复

设置昵称

关于作者

目录

加入云驻计划，成为创作者

推荐阅读

相关产品

方式 2：结合 Spring 的 `@Async`

方式 2：结合 Spring 的 `@Async`