多线程环境下订单ID重复问题的应用层解决方案需求

在多线程环境下调用pushOrderData方法批量处理从CSV读取的500条订单数据（含35+列），该方法逻辑为：查询订单表中最大OrderId，加1生成新OrderId，再将订单数据及关联数据保存至订单表及其他4-5张表。当前出现重复OrderId问题：多个线程读取到相同的最大OrderId，进而生成重复的新OrderId。已尝试用synchronized修饰pushOrderData方法，但问题未解决。

• 不依赖MySQL数据库的OrderId唯一约束；
• 需保留多线程处理的高性能（单线程处理需2.5-3分钟，多线程处理仅需10-12秒）；
• 寻求应用层解决方案。

多线程调用代码 #

List<Future<Void>> processFutures = new ArrayList<>();
for (BulkOrderData readData : bulkOrderDataList) {

    processFutures.add(processExcutorService.submit(() -> {
        try {
            PushOrderDataResponse placeOrderResponse = bulkPushOrderTransactionalService.pushOrderData(readData, productRefMap.get(readData.getItemData().get(0).getProductReference()), countryCode);
            readData.setOrderId(placeOrderResponse.getOrderId() != null ? String.valueOf(placeOrderResponse.getOrderId()) : "NA");
            readData.setUploadRemark(placeOrderResponse.getMessage());
        } catch (Exception e) {
            readData.setOrderId(null);
            readData.setUploadRemark("Error we got during placing this order, Please connect to technical team");
            logger.error("Error we got during push order in databse", e);
        }
        return null;
    }));
    // code....
}

数据保存类代码 #

// data saving class
@Service
public class BulkPushOrderTransactionalService {

    // method for saving 
    @Transactional(rollbackFor = Exception.class )
    public synchronized PushOrderDataResponse pushOrderData(BulkOrderData orderData, CassandraApiResponseReader cassandraApiResponseReader, String countryCode) throws Exception {
        int startOrderId = 100001;
        Integer resMaxOrderIdRes = 0;
        try {
            resMaxOrderIdRes = orderRepository.findOrderIdByOrderSource();
            // resMaxOrderIdRes = generateOrderId();
        } catch (Exception e) {
            throw new RuntimeException("Transaction failed in saving ", e);
        }

        if(resMaxOrderIdRes != null && resMaxOrderIdRes != 0) {
            startOrderId = resMaxOrderIdRes;
        }

        // new order id created by getting max order id firstly 
        newOrderId = startOrderId + 1;

        CustomerOrderDao customerOrderDao = new CustomerOrderDao();
        customerOrderDao.setOrderId(newOrderId);
        // more data setting

        /// saving data corrosponding to new order id 
        try{
            orderRepository.save(customerOrderDao);
            // orderDataService.saveOrder(customerOrderDao);
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("error in saving data in customer order table{}", e);
            response.setCode(404);
            response.setMessage("Error we got during placing this order, Please connect to technical team");
            response.setStatus("error");
            throw new RuntimeException("Transaction failed in saving customer item history", e);
        }

        // 4 more dao call where i am saving data
    }
}

1. 原synchronized失效原因 #

@Transactional会为方法生成代理类，synchronized修饰的是代理后的方法，但事务提交是在方法执行完毕后才完成的。线程A执行完synchronized方法后，事务还未提交，线程B进入方法查询最大OrderId时，读到的还是线程A提交前的旧值，因此会生成重复ID。

2. 可行应用层方案 #

方案一：预分配OrderId段（推荐，兼顾性能与正确性）

提前在应用层一次性获取一段连续的OrderId（比如一次拿500个），分发给各个线程使用，每个线程拿到的ID段不重叠，完全避免冲突：

• 实现步骤：
  1. 新增原子性方法allocateOrderIdRange(int count)，用synchronized（单实例）或分布式锁（集群）保证唯一执行，查询当前最大OrderId，返回[maxId+1, maxId+count]的区间，同时更新数据库中记录已分配最大ID的存储（比如新增一张ID分配表，或用订单表的一个占位记录）。
  2. 批量处理前调用该方法拿到500个ID的区间，循环给每个BulkOrderData分配唯一ID，线程直接用分配好的ID保存数据，不再查询最大ID。

方案二：原子类维护全局ID计数器

应用启动时读取数据库当前最大OrderId，用AtomicInteger维护计数器，线程通过getAndIncrement()获取唯一ID：

• 注意事项：
  1. 应用启动必须正确初始化原子类的值，确保与数据库最大ID一致。
  2. 应用重启时需重新读取数据库最大ID初始化原子类，避免ID重复。
  3. 单实例部署适用，集群环境需改用Redis的INCR命令实现分布式原子计数器。

方案三：调整锁范围覆盖事务提交

将synchronized移到事务外层，确保锁释放前事务已提交，避免线程读到未提交的旧数据：

@Service
public class BulkPushOrderTransactionalService {
    @Autowired
    private PlatformTransactionManager transactionManager;

    public PushOrderDataResponse pushOrderData(BulkOrderData orderData, CassandraApiResponseReader cassandraApiResponseReader, String countryCode) throws Exception {
        synchronized (this) {
            DefaultTransactionDefinition def = new DefaultTransactionDefinition();
            TransactionStatus status = transactionManager.getTransaction(def);
            try {
                int startOrderId = 100001;
                Integer resMaxOrderIdRes = orderRepository.findOrderIdByOrderSource();
                if(resMaxOrderIdRes != null && resMaxOrderIdRes != 0) {
                    startOrderId = resMaxOrderIdRes;
                }
                int newOrderId = startOrderId + 1;

                CustomerOrderDao customerOrderDao = new CustomerOrderDao();
                customerOrderDao.setOrderId(newOrderId);
                // 其他数据设置逻辑

                orderRepository.save(customerOrderDao);
                // 其他4个DAO保存逻辑

                transactionManager.commit(status);
                // 组装并返回响应
                PushOrderDataResponse response = new PushOrderDataResponse();
                response.setOrderId(newOrderId);
                response.setMessage("Success");
                response.setStatus("success");
                response.setCode(200);
                return response;
            } catch (Exception e) {
                transactionManager.rollback(status);
                logger.error("Transaction failed", e);
                PushOrderDataResponse response = new PushOrderDataResponse();
                response.setCode(404);
                response.setMessage("Error we got during placing this order, Please connect to technical team");
                response.setStatus("error");
                throw new RuntimeException("Transaction failed", e);
            }
        }
    }
}

• 缺点：此方案会让多线程串行执行，损失部分性能，适合无法实现预分配ID的保底场景。

内容的提问来源于stack exchange，提问作者Help out