Java多线程中AtomicInteger计数异常问题求助

AtomicInteger是否适用于该场景？ #

AtomicInteger本身完全适用于多线程环境下的计数场景，它的所有操作都是原子性的，不会出现线程安全问题。你遇到的计数残留问题，根源在于计数增减的逻辑存在原子性漏洞，或者异常处理遗漏，而非AtomicInteger本身不适用。

可能的问题原因 #

• URL检查与计数操作非原子化：你当前用Vector存储待处理URL，先检查URL是否已处理，再递增计数、添加Vector。但检查URL是否存在和递增计数+添加Vector是两个独立操作，多线程下可能出现竞态：两个线程同时检查到同一个URL未处理，都执行递增计数并添加到Vector，导致重复计数。后续其中一个任务处理完成递减计数，但另一个重复任务如果被跳过（比如处理时发现已处理），就会残留多余的计数。
• 计数递减逻辑未覆盖异常场景：如果Worker处理URL时抛出异常，且递减计数的代码没有放在finally块中，会导致该任务的计数没有被递减，最终总计数残留。
• 计数为0的判断非原子化：若你使用先获取当前计数、再递减、再判断的写法（比如if (taskCount.get() - 1 == 0) { taskCount.decrementAndGet(); }），这种拆分操作会导致竞态，多个线程同时递减后可能都错过计数为0的判断。

修复方案 #

方案一：修复AtomicInteger的使用逻辑 #

1. 原子化处理URL检查与计数
   用ConcurrentHashMap代替Vector记录已处理URL，利用putIfAbsent方法实现原子化的“检查+标记已处理”操作，避免重复计数：

   private final ConcurrentHashMap<String, Boolean> processedUrls = new ConcurrentHashMap<>();
   private final AtomicInteger taskCount = new AtomicInteger(0);
   
   public void submitUrl(String url) {
       // 原子化判断URL是否未处理，若未处理则标记并递增计数
       if (processedUrls.putIfAbsent(url, true) == null) {
           taskCount.incrementAndGet();
           executorService.submit(new ScrapperWorker(url, this));
       }
   }
   

2. 确保计数递减的可靠性
   将递减计数的代码放在finally块中，无论任务处理成功或失败都能执行：

   public class ScrapperWorker implements Runnable {
       private final String url;
       private final DoScrapper doScrapper;
   
       public ScrapperWorker(String url, DoScrapper doScrapper) {
           this.url = url;
           this.doScrapper = doScrapper;
       }
   
       @Override
       public void run() {
           try {
               // 处理当前URL，提取新的待处理URL
               List<String> newUrls = scrapeContent(url);
               // 提交新任务
               for (String newUrl : newUrls) {
                   doScrapper.submitUrl(newUrl);
               }
           } catch (Exception e) {
               // 异常日志记录
               e.printStackTrace();
           } finally {
               // 原子化递减计数并判断是否为0
               int remainingTasks = taskCount.decrementAndGet();
               if (remainingTasks == 0) {
                   // 设置实体状态为DONE并持久化
                   doScrapper.markEntityAsDone();
               }
           }
       }
   
       private List<String> scrapeContent(String url) {
           // 爬虫逻辑实现
           return new ArrayList<>();
       }
   }
   

方案二：使用Phaser处理动态任务（更优雅的替代方案） #

如果你的爬虫任务是动态生成的（处理一个URL可能生成多个新任务），Phaser比AtomicInteger更适合，它支持动态注册任务，能自动跟踪所有任务的完成状态：

private final Phaser phaser = new Phaser(1); // 初始注册1个主任务
private final ConcurrentHashMap<String, Boolean> processedUrls = new ConcurrentHashMap<>();

public void startScrape() {
    // 提交初始任务
    submitUrl(initialUrl);
    // 等待所有任务完成
    phaser.awaitAdvance(phaser.getPhase());
    // 标记实体为DONE并持久化
    markEntityAsDone();
}

public void submitUrl(String url) {
    if (processedUrls.putIfAbsent(url, true) == null) {
        phaser.register(); // 注册新任务
        executorService.submit(() -> {
            try {
                List<String> newUrls = scrapeContent(url);
                for (String newUrl : newUrls) {
                    submitUrl(newUrl);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                phaser.arrive(); // 任务完成，通知Phaser
            }
        });
    }
}

Phaser的优势在于无需手动维护计数，它会自动管理所有动态注册的任务，当所有任务完成时，awaitAdvance会自动解除阻塞，触发后续的状态设置逻辑。

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