任意键的锁定处理程序

Question

我有为任意键实现“锁定处理程序”的代码。给定一个key，它确保一次只有一个线程可以process那个（或等于）键（这里意味着调用externalSystem.process(key)调用）。

到目前为止，我有这样的代码：

public class MyHandler {
    private final SomeWorkExecutor someWorkExecutor;
    private final ConcurrentHashMap<Key, Lock> lockMap = new ConcurrentHashMap<>();

    public void handle(Key key) {
        // This can lead to OOM as it creates locks without removing them
        Lock keyLock = lockMap.computeIfAbsent( 
            key, (k) -> new ReentrantLock()
        );
        keyLock.lock();
        try {
            someWorkExecutor.process(key);
        } finally {
            keyLock.unlock();
        }
    }
}

我知道这段代码可以导致OutOfMemoryError，因为没有人清楚地图。

我在考虑如何制作能够累积有限元素数量的地图。当超出限制时，我们应该用新的替换最旧的访问元素（此代码应与最旧的元素同步作为监视器）。但我不知道如何让回调告诉我超出限制。

请分享你的想法。

附言

我重读了任务，现在我发现我有限制，handle 方法不能被调用超过 8 个线程。我不知道它对我有什么帮助，但我只是提到了它。

附言 2

@Boris the Spider 提出了不错且简单的解决方案：

} finally {
      lockMap.remove(key);
      keyLock.unlock();
}

但是在 Boris 注意到我们的代码不是线程安全的，因为它破坏了行为之后：
让我们研究 3 个使用相同键调用的线程：

1. Thread#1 获取锁，现在在 map.remove(key); 之前
2. Thread#2 使用 equals 键调用，因此它在 thread#1 释放锁定时等待。
3. 然后线程#1 执行map.remove(key);。在这个线程#3 调用方法handle 之后。它检查该键的锁是否在映射中不存在，因此它创建新锁并获取它。
4. 线程#1 释放锁，因此线程#2 获取它。
   因此，线程#2 和线程#3 可以并行调用等于键。但这不应该被允许。

为了避免这种情况，在map清除之前，我们应该阻塞任何线程来获取锁，而waitset中的所有线程都没有获取并释放锁。看起来它需要足够复杂的同步，它会导致算法工作缓慢。当地图大小超过某个限制值时，也许我们应该不时清除地图。

我浪费了很多时间，但不幸的是我不知道如何实现这一点。

Answer 1

您无需尝试将大小限制为某个任意值 - 事实证明，您可以完成这种“锁定处理程序”习语，同时只存储准确键的数量目前锁定在地图中。

这个想法是使用一个简单的约定：成功地添加映射到映射算作“锁定”操作，删除它算作“解锁”操作。这巧妙地避免了在某些线程仍处于锁定状态和其他竞争条件时删除映射的问题。

此时，映射中的value只用于阻塞其他以相同key到达的线程，需要等到映射被移除。

这是一个示例¹，地图值使用CountDownLatch 而不是Lock：

public void handle(Key key) throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(1);

    // try to acquire the lock by inserting our latch as a
    // mapping for key        
    while(true) {
        CountDownLatch existing = lockMap.putIfAbsent(key, latch);
        if (existing != null) {
            // there is an existing key, wait on it
            existing.await();
        } else {
            break;
        }
    }

    try {
        externalSystem.process(key);
    } finally {
        lockMap.remove(key);
        latch.countDown();
    }
}

在这里，映射的生命周期仅与持有锁的时间一样长。映射的条目永远不会超过对不同键的并发请求。

您的方法的不同之处在于映射不会“重复使用” - 每个handle 调用都会创建一个新的锁存器和映射。由于您已经在执行昂贵的原子操作，因此在实践中这不太可能会放缓。另一个缺点是，由于有许多等待线程，当锁存器倒计时时，所有都会被唤醒，但只有一个会成功放入新的映射并因此获得锁——其余的会在新锁。

您可以构建另一个版本，当线程出现并等待现有映射时，它会重新使用映射。基本上，解锁线程只是对其中一个等待线程进行“切换”。只有一个映射将用于等待同一键的整个线程集 - 它按顺序传递给每个线程。大小仍然是有界的，因为没有更多线程在等待给定的映射，它仍然被删除。

为了实现这一点，您将CountDownLatch 替换为可以计算等待线程数的映射值。当一个线程进行解锁时，它首先检查是否有线程在等待，如果是，则唤醒一个进行切换。如果没有线程在等待，它会“销毁”对象（即，设置一个标志，表明该对象不再在映射中）并将其从映射中删除。

您需要在适当的锁下进行上述操作，并且有一些棘手的细节。在实践中，我发现上面这个简短而甜蜜的例子效果很好。

---

¹ 即时编写，未编译，未测试，但想法可行。

Answer 2

您可以依靠方法 compute(K key, BiFunction<? super K,? super V,? extends V> remappingFunction) 来同步对给定键的方法 process 的调用，您甚至不再需要使用 Lock 作为地图值的类型，因为您不需要不要再依赖它了。

这个想法是依靠 ConcurrentHashMap 的内部锁定机制来执行你的方法，这将允许线程并行执行 process 方法，用于对应哈希值不属于同一 bin 的键。这相当于基于条带锁的方法，只是不需要额外的第三方库。

条带锁的方法很有趣，因为它在内存占用方面非常轻，因为您只需要有限数量的锁来执行此操作，因此您的锁所需的内存占用是已知的并且永远不会改变，这不是对于每个键使用一个锁的方法（如您的问题），因此通常更好/建议使用基于条带锁的方法来满足这种需求。

所以你的代码可能是这样的：

// This will create a ConcurrentHashMap with an initial table size of 16   
// bins by default, you may provide an initialCapacity and loadFactor
// if too much or not enough to get the expected table size in order
// increase or reduce the concurrency level of your map
// NB: We don't care much of the type of the value so I arbitrarily
// used Void but it could be any type like simply Object
private final ConcurrentMap<Key, Void> lockMap = new ConcurrentHashMap<>();

public void handle(Key lockKey) {
    // Execute the method process through the remapping Function
    lockMap.compute(
        lockKey,
        (key, value) -> {
            // Execute the process method under the protection of the
            // lock of the bin of hashes corresponding to the key
            someWorkExecutor.process(key);
            // Returns null to keep the Map empty
            return null;
        }
    );
}

NB 1：因为我们总是返回null，所以映射总是空的，这样你就永远不会因为这个映射而耗尽内存。

注意 2：由于我们从不影响给定键的值，请注意，也可以使用方法 computeIfAbsent(K key, Function<? super K,? extends V> mappingFunction)：

public void handle(Key lockKey) {
    // Execute the method process through the remapping Function
    lockMap.computeIfAbsent(
        lockKey,
        key -> {
            // Execute the process method under the protection of the
            // lock of the segment of hashes corresponding to the key
            someWorkExecutor.process(key);
            // Returns null to keep the Map empty
            return null;
        }
    );
}

注意 3： 确保您的方法 process 永远不会为任何键调用方法 handle，因为您最终会出现无限循环（相同的键）或死锁（其他非有序键） ，例如：如果一个线程调用handle(key1)，然后process内部调用handle(key2)，另一个线程并行调用handle(key2)，然后process内部调用handle(key1)，无论使用哪种方法，都会出现死锁）。此行为并非特定于此方法，它会发生在任何方法中。

Answer 3

一种方法是完全放弃并发哈希映射，而只需使用带有锁定功能的常规 HashMap 来执行所需的映射操作并以原子方式锁定状态。

乍一看，这似乎降低了系统的并发性，但如果我们假设process(key) 调用相对于非常快速的锁操作来说是冗长的，那么它工作得很好，因为process() 调用仍然并发运行。独占临界区只发生少量且固定的工作。

这是一个草图：

public class MyHandler {

    private static class LockHolder {
        ReentrantLock lock = new ReentrantLock();
        int refcount = 0;
        void lock(){
            lock.lock();
        }
    } 

    private final SomeWorkExecutor someWorkExecutor;
    private final Lock mapLock = new ReentrantLock();
    private final HashMap<Key, LockHolder> lockMap = new HashMap<>();

    public void handle(Key key) {

        // lock the map
        mapLock.lock();
        LockHolder holder = lockMap.computeIfAbsent(key, k -> new LockHolder());
        // the lock in holder is either unlocked (newly created by us), or an existing lock, let's increment refcount
        holder.refcount++;
        mapLock.unlock();

        holder.lock();

        try {
            someWorkExecutor.process(key);
        } finally {
            mapLock.lock()
            keyLock.unlock();
            if (--holder.refcount == 0) {
              // no more users, remove lock holder
              map.remove(key);
            }
            mapLock.unlock();
        }
    }
}

我们使用refcount，它只在共享的mapLock 下进行操作，以跟踪有多少个锁的用户。每当引用计数为零时，我们可以在退出处理程序时摆脱条目。这种方法很好，因为它很容易推理并且如果process() 调用与锁定开销相比相对昂贵，它会很好地执行。由于地图操作发生在共享锁下，因此添加其他逻辑也很简单，例如，在地图中保留一些 Holder 对象、跟踪统计信息等。

Answer 4

谢谢Ben Mane
我找到了这个变种。

public class MyHandler {
    private final int THREAD_COUNT = 8;
    private final int K = 100;
    private final Striped<Lock> striped = Striped.lazyWeakLock(THREAD_COUNT * K);
    private final SomeWorkExecutor someWorkExecutor = new SomeWorkExecutor();

    public void handle(Key key) throws InterruptedException {
        Lock keyLock = striped.get(key);

        keyLock.lock();
        try {
            someWorkExecutor.process(key);
        } finally {
            keyLock.unlock();
        }       
    }
}

Answer 5

这是一个简短而甜蜜的版本，它利用 Guava 的 Interner 类的 weak 版本来完成为每个键提供“规范”对象以用作锁的繁重工作，并实现弱引用语义以便清理未使用的条目。

public class InternerHandler {
    private final Interner = Interners.newWeakInterner();

    public void handle(Key key) throws InterruptedException {
        Key canonKey = Interner.intern(key);
        synchronized (canonKey) {
            someWorkExecutor.process(key);
        }       
    }
}

基本上我们要求一个规范canonKey，即equal() 到key，然后锁定这个canonKey。每个人都会就规范密钥达成一致，因此所有传递相同密钥的调用者都会就锁定对象达成一致。

Interner 的弱特性意味着只要不使用规范密钥，就可以删除条目，因此您可以避免在 interner 中积累条目。稍后，如果再次出现相等键，则选择新的规范条目。

上面的简单代码依赖于 synchronize 的内置监视器 - 但如果这对您不起作用（例如，它已经用于其他目的），您可以在 Key 类中包含一个锁定对象或者创建一个持有者对象。

Answer 6

class MyHandler {
    private final Map<Key, Lock> lockMap = Collections.synchronizedMap(new WeakHashMap<>());
    private final SomeWorkExecutor someWorkExecutor = new SomeWorkExecutor();

    public void handle(Key key) throws InterruptedException {
        Lock keyLock = lockMap.computeIfAbsent(key, (k) -> new ReentrantLock()); 
        keyLock.lock();
        try {
            someWorkExecutor.process(key);
        } finally {
            keyLock.unlock();
        }
    }
}

Answer 7

就性能而言，每次为key 创建和删除锁定对象是一项代价高昂的操作。当您从并发映射（例如缓存）中添加/删除锁时，必须确保从缓存中放置/删除对象本身是线程安全的。所以这似乎不是一个好主意，但可以通过ConcurrentHashMap 实现

条带锁定方法（也被并发散列映射内部使用）是更好的方法。从Google Guava docs 解释为

当你想将一个锁与一个对象关联时，key 保证 你需要的是，如果 key1.equals(key2)，那么与关联的锁 key1 与 key2 关联的锁相同。

最粗略的方法是将每个键与相同的键关联 锁，这会导致最粗略的同步。在 另一方面，您可以将每个不同的键与不同的键相关联 锁，但这需要线性内存消耗和并发 随着新钥匙的发现，对锁系统本身进行管理。

条带化允许程序员选择多个锁，它们是 基于它们的哈希码在键之间分布。这允许 程序员动态选择并发和之间的权衡 内存消耗，同时保留如果 key1.equals(key2)，然后 striped.get(key1) == striped.get(key2)

代码：

//declare globally; e.g. class field level
Striped<Lock> rwLockStripes = Striped.lock(16);

    Lock lock = rwLockStripes.get("key");
    lock.lock();
    try {
        // do you work here
    } finally {
        lock.unlock();
    }

以下代码片段有助于实现锁定/解除锁定。

private ConcurrentHashMap<String, ReentrantLock> caches = new ConcurrentHashMap<>();

public void processWithLock(String key) {
    ReentrantLock lock = findAndGetLock(key);
    lock.lock();
    try {
        // do you work here

    } finally {
        unlockAndClear(key, lock);
    }
}

private void unlockAndClear(String key, ReentrantLock lock) {
    // *** Step 1: Release the lock.
    lock.unlock();
    // *** Step 2: Attempt to remove the lock
    // This is done by calling compute method, if given lock is present in
    // cache. if current lock object in cache is same instance as 'lock'
    // then remove it from cache. If not, some other thread is succeeded in
    // putting new lock object and hence we can leave the removal of lock object to that
    // thread.
    caches.computeIfPresent(key, (k, current) -> lock == current ? null : current);

}

private ReentrantLock findAndGetLock(String key) {
    // Merge method given us the access to the previously( if available) and
    // newer lock object together.
    return caches.merge(key, new ReentrantLock(), (older, newer) -> nonNull(older) ? older : newer);
}

Answer 8

您可以尝试JKeyLockManager 之类的东西，而不是写你自己的。来自项目描述：

JKeyLockManager 为应用程序提供细粒度锁定 特定的键。

现场给出的示例代码：

public class WeatherServiceProxy {
  private final KeyLockManager lockManager = KeyLockManagers.newManager();

  public void updateWeatherData(String cityName, float temperature) {
    lockManager.executeLocked(cityName, () -> delegate.updateWeatherData(cityName, temperature)); 
  }

Answer 9

调用时会添加新值

lockMap.computeIfAbsent()

所以你可以只检查 lockMap.size() 的项目数。

但是您将如何找到第一个添加的项目？最好在使用后删除项目。

Answer 10

您可以使用存储对象引用的进程内缓存，例如 Caffeine、Guava、EHCache 或 cache2k。下面是一个如何使用cache2k 构建缓存的示例：

final Cache<Key, Lock> locks =
  new Cache2kBuilder<Key, Lock>(){}
    .loader(
      new CacheLoader<Key, Lock>() {
        @Override
        public Lock load(Key o) {
          return new ReentrantLock();
        }
      }
    )
    .storeByReference(true)
    .entryCapacity(1000)
    .build();

使用模式与您在问题中的情况一样：

    Lock keyLock = locks.get(key);
    keyLock.lock();
    try {
        externalSystem.process(key);
    } finally {
        keyLock.unlock();
    }

由于缓存限制为 1000 个条目，因此会自动清除不再使用的锁。

如果应用程序中的容量和线程数不匹配，缓存可能会清除正在使用的锁。该解决方案在我们的应用中可以完美运行多年。当有足够长的运行任务并且超出容量时，缓存将驱逐正在使用的锁。在实际应用程序中，您始终控制生命线程的数量，例如在 Web 容器中，您会将处理线程的数量限制为（示例）100。因此您知道使用的锁永远不会超过 100 个。如果考虑到这一点，则此解决方案的开销最小。

请记住，只有当您的应用程序在单个 VM 上运行时，锁定才有效。您可能想看看分布式锁管理器 (DLM)。提供分布式锁的产品示例：hazelcast、infinispan、teracotta、redis/redisson。