使用Async方法时遭遇Java堆内存溢出错误排查

问题背景 #

在调用异步方法并使用ThreadPoolExecutorService时，遇到java.lang.OutOfMemoryError: Java heap space错误，已尝试移除executor.setQueueCapacity(10);但问题仍存在，相关代码如下：

线程池配置Bean #

@Bean(name = "taskExecutor")
public ThreadPoolTaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(5);
    executor.setQueueCapacity(10);
    executor.setThreadNamePrefix("SyncThread");
    executor.initialize();
    return executor;
}

异步请求接口 #

@RequestMapping(value = "/coupons/bulk/syncCoupons", method = RequestMethod.POST)
@Async("taskExecutor")
public ResponseEntity<?> syncShopifyCouponsV4(@RequestBody List<Long> ids) {
    logger.info("Start :: syncShopifyPriceRulesListToMongoV3 for ids size = " + ids.size());
    try {
        if (!ids.isEmpty()) {
            List<CompletableFuture<Void>> futures = ids.stream()
                    .map(id -> CompletableFuture.runAsync(() -> {
                        logger.info("id inside CompletableFuture: " + id);
                        ShopifyBulkSyncCoupons task = shopifyBulkSyncCoupons;
                        task.setCouponId(id);
                        task.run(); // Manually invoke the run method
                    }, taskExecutor)) // Use your configured taskExecutor here
                    .collect(Collectors.toList());

            CompletableFuture<Void>[] futureArray = futures.toArray(new CompletableFuture[0]);
            CompletableFuture<Void> allOf = CompletableFuture.allOf(futureArray);
            allOf.join(); // Wait for all tasks to complete
        }
    }
    catch (Exception e) {
        throw new RuntimeException(e);
    }

    logger.info("End :: syncShopifyPriceRulesListToMongoV3");
    return new ResponseEntity<>(HttpStatus.OK);
}

同步业务方法 #

public void syncShopifyPriceRulesListToMongoV2( Long id)
{
    ShopifyClient shopifyClient1 = new ShopifyClient();
    ShopifyPriceRuleResponse priceRuleResponse = shopifyClient1.getPriceRuleById(id);
    if(null!=priceRuleResponse && null!=priceRuleResponse.getPriceRule()) {
        PriceRuleDetails priceRuleDetails = new PriceRuleDetails();
        List<Long> finalEntitledVariants = new ArrayList<>();
        priceRuleDetails = mapShopifyPriceRuleToModel(priceRuleResponse.getPriceRule());
        if (!CollectionUtils.isEmpty(priceRuleDetails.getEntitledCollectionIds())) {
            List<Long> collectionIds = priceRuleDetails.getEntitledCollectionIds();
            List<ProductAttributes> prodAttr = productAttrRepository.findByCollectionsIn(collectionIds);
            for (ProductAttributes prod : prodAttr) {
                finalEntitledVariants.add(prod.getVariantId());
            }
        }
        if (!CollectionUtils.isEmpty(priceRuleDetails.getEntitledProductIds())) {
            List<ProductAttributes> productAttributesList = productAttrRepository.findByProductIdIn(priceRuleDetails.getEntitledProductIds());
            for (ProductAttributes prod : productAttributesList) {
                finalEntitledVariants.add(prod.getVariantId());
            }
        }
        if (!CollectionUtils.isEmpty(priceRuleDetails.getEntitledVariantIds())) {
            finalEntitledVariants.addAll(priceRuleDetails.getEntitledVariantIds());
        }
        priceRuleDetails.setEntitledVariantIds(finalEntitledVariants);
        priceRuleDetailsRepo.save(priceRuleDetails);
    }

}

Runnable任务类 #

public class ShopifyBulkSyncCoupons implements Runnable
{

    private static final Logger logger = 
    Logger.getLogger(PriceRuleServiceImpl.class);
    private  Long couponId ;

    public ShopifyBulkSyncCoupons(Long couponId) {
        this.couponId = couponId;
    }

    public Long getCouponId() {
        return couponId;
    }

    public void setCouponId(Long couponId) {
        couponId = couponId;
    }

    @Override
    @Async("taskExecutor")
    public void run()
    {
        logger.info("Inside run for id "+couponId+" Thread is "+Thread.currentThread().getName());
         PriceRuleServiceImpl priceRuleService = new PriceRuleServiceImpl();
         priceRuleService.syncShopifyPriceRulesListToMongoV2(couponId);
    }
}

问题根源分析 #

• 三重异步嵌套导致任务爆炸
  接口方法syncShopifyCouponsV4已标记@Async，会被提交到线程池；内部又用CompletableFuture.runAsync再次提交任务到同一线程池；Runnable类的run方法又加了@Async。当传入的ids数量大时，会瞬间产生大量待执行任务，内存被任务对象、线程栈等占满。

• Runnable类的致命bug
  setCouponId方法中couponId = couponId;是局部变量赋值，根本没修改成员变量，导致所有任务的couponId值错误；同时手动new PriceRuleServiceImpl()会绕过Spring依赖注入，每次执行都创建新的Service实例，大量实例堆积占用堆内存。

• 无限制任务提交
  不管ids规模多大，一次性提交所有任务，没有分批或限流，瞬间生成大量CompletableFuture对象和任务，内存占用急剧上升。

• 业务方法内存隐患
  业务方法中查询ProductAttributes时，若返回数据量极大（比如单collection对应上万条产品），多个并发任务同时加载大量数据，直接压垮堆内存。

修复方案 #

1. 移除多余异步嵌套，简化任务提交 #

去掉重复的@Async注解，避免多层异步调用。示例修改后的接口：

@RequestMapping(value = "/coupons/bulk/syncCoupons", method = RequestMethod.POST)
public ResponseEntity<?> syncShopifyCouponsV4(@RequestBody List<Long> ids) {
    logger.info("Start :: syncShopifyPriceRulesListToMongoV3 for ids size = " + ids.size());
    try {
        if (!ids.isEmpty()) {
            // 用Semaphore限制同时执行的任务数
            Semaphore semaphore = new Semaphore(10);
            ids.forEach(id -> {
                taskExecutor.submit(() -> {
                    try {
                        semaphore.acquire();
                        logger.info("id inside task: " + id);
                        priceRuleService.syncShopifyPriceRulesListToMongoV2(id);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    } finally {
                        semaphore.release();
                    }
                });
            });
        }
    } catch (Exception e) {
        throw new RuntimeException(e);
    }

    logger.info("End :: syncShopifyPriceRulesListToMongoV3");
    return new ResponseEntity<>(HttpStatus.OK);
}

2. 修复Runnable类bug，改用依赖注入 #

• 修复setCouponId方法：this.couponId = couponId;
• 不要手动创建Service实例，将ShopifyBulkSyncCoupons交给Spring管理，注入PriceRuleServiceImpl；或者直接省略Runnable类，在任务中直接调用Service方法。

3. 分批处理+合理配置线程池 #

• 对ids进行分批处理，比如每50个id为一批，避免一次性提交大量任务；
• 调整线程池参数：根据服务器CPU核心数设置corePoolSize（比如CPU核心数2），maxPoolSize可设为核心数4，queueCapacity设为100-500，同时设置任务拒绝策略。

4. 优化业务方法内存占用 #

• 对productAttrRepository的查询改为分页查询，避免一次性加载大量数据：

  // 示例分页查询逻辑
  int page = 0;
  int pageSize = 100;
  List<ProductAttributes> prodAttr;
  do {
      Pageable pageable = PageRequest.of(page++, pageSize);
      prodAttr = productAttrRepository.findByCollectionsIn(collectionIds, pageable);
      for (ProductAttributes prod : prodAttr) {
          finalEntitledVariants.add(prod.getVariantId());
      }
  } while (!prodAttr.isEmpty());
  

• 用流式处理替代循环，减少内存占用。

内容的提问来源于stack exchange，提问作者Shreya Aggarwal