Java基于ConcurrentMap实现键粒度锁与空闲锁清理的并发问题求解

你当前的实现存在锁对象生命周期的竞态条件，具体触发场景如下：

• 线程A首次创建key对应的锁lock1存入锁映射表，获取锁1进入临界区
• 线程B后续调用putIfAbsent拿到锁1，阻塞等待线程A释放锁
• 线程A执行完临界区后，执行LOCKS.remove(key)将锁1从映射表移除
• 线程C此时调用putIfAbsent，发现映射表中无对应key，创建新锁lock2存入并直接获取锁2执行临界区
• 线程A释放锁1后，线程B拿到锁1开始执行临界区

此时线程B（锁1）和线程C（锁2）同时进入临界区，直接触发断言失败。加全局锁能解决是因为串行化了所有锁的获取/删除流程，消除了上述竞态。

使用带引用计数的自定义锁对象，保证锁只有在完全没有线程持有时才会被从映射表移除，完全不需要全局大锁。
具体实现逻辑：

• 自定义带引用计数的锁类，封装同步用的monitor和当前持锁/等锁的线程计数
• 用ConcurrentHashMap.computeIfAbsent原子获取/创建对应key的锁对象，避免多线程创建多个同key锁
• 每次获取锁前先给引用计数+1，用完释放锁后给引用计数-1
• 只有当引用计数归0时，才尝试删除映射表中的锁（用remove(key, value)的重载避免误删新创建的锁）

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.testng.Assert;

public class FixedIssue {
    private static final int THREAD_COUNT = 1000;
    private static final ConcurrentMap<String, RefCountedLock> LOCKS = new ConcurrentHashMap<>();
    private static final AtomicBoolean SOME_SCOPED_RESOURCE = new AtomicBoolean(false);

    // 带引用计数的自定义锁
    static final class RefCountedLock {
        final AtomicInteger refCount = new AtomicInteger(0);
        // 实际用于同步的monitor对象
        final Object monitor = new Object();
    }

    public static void main(String[] args) throws Throwable {
        ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
        CompletionService<Boolean> completionService = new ExecutorCompletionService<>(threadPool);
        AtomicReference<Throwable> taskException = new AtomicReference<>();

        for (int currentThreadNumber = 0; currentThreadNumber < THREAD_COUNT; currentThreadNumber++) {
            final int taskNumber = currentThreadNumber;
            completionService.submit(() -> {
                try {
                    someTask("someKey", taskNumber);
                } catch (Throwable t) {
                    t.printStackTrace();
                    taskException.set(t);
                }
                return true;
            });
        }

        for (int currentThreadNumber = 0; currentThreadNumber < THREAD_COUNT; currentThreadNumber++) {
            completionService.take();
        }
        threadPool.shutdownNow();

        if (taskException.get() != null) {
            throw taskException.get();
        }
    }

    public static void someTask(String key, int taskNumber) {
        // 原子获取/创建对应key的锁，computeIfAbsent保证同key只会创建一个锁对象
        RefCountedLock lock = LOCKS.computeIfAbsent(key, k -> new RefCountedLock());
        // 给引用计数+1，标记当前有线程使用该锁
        lock.refCount.incrementAndGet();

        try {
            // 用实际的monitor执行同步
            synchronized (lock.monitor) {
                System.err.println(taskNumber + ": lock=" + lock.monitor);
                Assert.assertFalse(SOME_SCOPED_RESOURCE.getAndSet(true), "failed for task " + taskNumber);
                System.err.println("did some work for task " + taskNumber);
                Assert.assertTrue(SOME_SCOPED_RESOURCE.getAndSet(false), "failed for task " + taskNumber);
            }
        } finally {
            // 释放锁后给引用计数-1，用finally保证异常场景下也能正确递减
            int remainingRef = lock.refCount.decrementAndGet();
            // 只有引用计数归0时，才尝试删除映射表中的锁，且只有映射表中存的还是当前锁时才删
            if (remainingRef == 0) {
                LOCKS.remove(key, lock);
            }
        }
    }
}

• ConcurrentHashMap.computeIfAbsent对同一个key的计算是原子的，天然避免了多线程创建多个同key锁的问题，不需要额外加锁
• LOCKS.remove(key, lock)的双参重载是关键：如果其他线程已经在你释放锁的间隙创建了新锁存入映射表，该操作不会误删新锁
• 不同key的锁操作完全并行，只有同key的操作会串行，性能远高于全局锁方案

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