我正在尝试使用RentrantLock实现一个类来强制执行我的Java应用程序中的并发性,阻止异步修改给定实体实例(由关键字标识)。其目标是阻塞/排队多个并发修改给定对象实例的尝试,直到先前的线程完成。该类以通用方式实现了这一点,允许任何代码块获取锁并在完成后释放它(与RentrantLock语义相同),同时增加了仅阻塞试图修改对象的同一实例的线程(根据关键字标识)而不是阻塞进入代码块的所有线程的效用。
该类提供了一个简单的结构,只允许为一个实体实例同步一个代码块。例如,如果我希望代码块对来自ID为33的用户的所有线程进行同步,但我不希望来自任何其他用户的线程被服务用户33的线程阻塞。
该类实现如下:
问题在于它不能完美地工作。在非常高的并发负载下,仍然存在一些情况,其中具有相同“key”的多个线程没有被同步。我已经进行了彻底的测试,但无法弄清楚为什么/在哪里会出现这种情况。
有任何并发专家吗?
该类提供了一个简单的结构,只允许为一个实体实例同步一个代码块。例如,如果我希望代码块对来自ID为33的用户的所有线程进行同步,但我不希望来自任何其他用户的线程被服务用户33的线程阻塞。
该类实现如下:
public class EntitySynchronizer {
private static final int DEFAULT_MAXIMUM_LOCK_DURATION_SECONDS = 300; // 5 minutes
private Object mutex = new Object();
private ConcurrentHashMap<Object, ReentrantLock> locks = new ConcurrentHashMap<Object, ReentrantLock>();
private static ThreadLocal<Object> keyThreadLocal = new ThreadLocal<Object>();
private int maximumLockDurationSeconds;
public EntitySynchronizer() {
this(DEFAULT_MAXIMUM_LOCK_DURATION_SECONDS);
}
public EntitySynchronizer(int maximumLockDurationSeconds) {
this.maximumLockDurationSeconds = maximumLockDurationSeconds;
}
/**
* Initiate a lock for all threads with this key value
* @param key the instance identifier for concurrency synchronization
*/
public void lock(Object key) {
if (key == null) {
return;
}
/*
* returns the existing lock for specified key, or null if there was no existing lock for the
* key
*/
ReentrantLock lock;
synchronized (mutex) {
lock = locks.putIfAbsent(key, new ReentrantLock(true));
if (lock == null) {
lock = locks.get(key);
}
}
/*
* Acquires the lock and returns immediately with the value true if it is not held by another
* thread within the given waiting time and the current thread has not been interrupted. If this
* lock has been set to use a fair ordering policy then an available lock will NOT be acquired
* if any other threads are waiting for the lock. If the current thread already holds this lock
* then the hold count is incremented by one and the method returns true. If the lock is held by
* another thread then the current thread becomes disabled for thread scheduling purposes and
* lies dormant until one of three things happens: - The lock is acquired by the current thread;
* or - Some other thread interrupts the current thread; or - The specified waiting time elapses
*/
try {
/*
* using tryLock(timeout) instead of lock() to prevent deadlock situation in case acquired
* lock is not released normalRelease will be false if the lock was released because the
* timeout expired
*/
boolean normalRelease = lock.tryLock(maximumLockDurationSeconds, TimeUnit.SECONDS);
/*
* lock was release because timeout expired. We do not want to proceed, we should throw a
* concurrency exception for waiting thread
*/
if (!normalRelease) {
throw new ConcurrentModificationException(
"Entity synchronization concurrency lock timeout expired for item key: " + key);
}
} catch (InterruptedException e) {
throw new IllegalStateException("Entity synchronization interrupted exception for item key: "
+ key);
}
keyThreadLocal.set(key);
}
/**
* Unlock this thread's lock. This takes care of preserving the lock for any waiting threads with
* the same entity key
*/
public void unlock() {
Object key = keyThreadLocal.get();
keyThreadLocal.remove();
if (key != null) {
ReentrantLock lock = locks.get(key);
if (lock != null) {
try {
synchronized (mutex) {
if (!lock.hasQueuedThreads()) {
locks.remove(key);
}
}
} finally {
lock.unlock();
}
} else {
synchronized (mutex) {
locks.remove(key);
}
}
}
}
}
这个类的使用方法如下:
private EntitySynchronizer entitySynchronizer = new EntitySynchronizer();
entitySynchronizer.lock(userId); // 'user' is the entity by which i want to differentiate my synchronization
try {
//execute my code here ...
} finally {
entitySynchronizer.unlock();
}
问题在于它不能完美地工作。在非常高的并发负载下,仍然存在一些情况,其中具有相同“key”的多个线程没有被同步。我已经进行了彻底的测试,但无法弄清楚为什么/在哪里会出现这种情况。
有任何并发专家吗?
entitySynchronizer
实例? - assylias