Java中如何为Executor Thread分组执行线程？实现单账户串行处理

针对你的需求，核心思路是为每个bankAccountId分配独立的锁资源，让同一Id的任务串行执行，不同Id的任务可以借助线程池并行处理。下面是具体实现方案：

核心实现逻辑 #

利用ConcurrentHashMap线程安全的特性，为每个bankAccountId维护唯一的锁对象。当任务提交到线程池后，先获取对应Id的锁，再执行具体业务逻辑，确保同一Id的任务不会被多线程同时处理。

1. 账户锁管理器类 #

负责为每个bankAccountId创建并维护唯一的锁对象：

import java.util.concurrent.ConcurrentHashMap;

public class AccountLockManager {
    private final ConcurrentHashMap<String, Object> accountLocks = new ConcurrentHashMap<>();

    // 线程安全地获取对应账户的锁，不存在则自动创建
    public Object getLock(String bankAccountId) {
        return accountLocks.computeIfAbsent(bankAccountId, key -> new Object());
    }
}

2. 结合ExecutorService的任务处理类 #

将锁逻辑嵌入任务内部，提交到线程池执行：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class AccountProcessor {
    // 初始化固定大小的线程池
    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
    private final AccountLockManager lockManager = new AccountLockManager();

    public void processAccount(String bankAccountId) {
        executorService.submit(() -> {
            // 获取当前账户对应的锁
            Object accountLock = lockManager.getLock(bankAccountId);
            
            // 加锁执行业务逻辑
            synchronized (accountLock) {
                // 替换为你的账户处理逻辑（如转账、余额更新等）
                System.out.printf("线程[%s]开始处理账户[%s]%n", 
                    Thread.currentThread().getName(), bankAccountId);
                
                // 模拟业务处理耗时
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                
                System.out.printf("线程[%s]完成账户[%s]处理%n", 
                    Thread.currentThread().getName(), bankAccountId);
            }
        });
    }

    // 程序退出时关闭线程池
    public void shutdown() {
        executorService.shutdown();
    }
}

方案优势 #

• 资源利用率高：不同bankAccountId的任务可并行执行，充分发挥线程池的性能
• 线程安全：ConcurrentHashMap的computeIfAbsent方法保证同一Id只会生成一个锁对象，避免并发创建问题
• 实现简洁：无需复杂的线程组或信号量配置，用基础的synchronized即可满足排他性需求

注意事项 #

• 如果业务中存在大量过期的bankAccountId，可定期清理ConcurrentHashMap中的锁对象（比如配合弱引用或定时任务），避免内存占用过高
• 不要在提交任务前获取锁，否则会阻塞线程池的任务提交流程，浪费线程资源

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