java - 如何同步多个线程访问一些公共(public)数据

Question

我有三个不同的线程，它们创建三个不同的对象来读取/操作所有线程通用的一些数据。现在，我需要确保一次仅向一个线程授予访问权限。

这个例子是这样的。

public interface CommonData {
   public void addData(); // adds data to the cache
   public void getDataAccessKey(); // Key that will be common across different threads for each data type
}

/*
* Singleton class
*/
public class CommonDataCache() {
   private final Map dataMap = new HashMap(); // this takes keys and values as custom objects
}

接口(interface)的实现类如下所示

class CommonDataImpl implements CommonData {
    private String key;
    public CommonDataImpl1(String key) {
       this.key = key;
    }

    public void addData() {
      // access the singleton cache class and add
    }

    public void getDataAccessKey() {
      return key;
    }
}

每个线程将按如下方式调用:

CommonData data = new CommonDataImpl("Key1");
new Thread(() -> data.addData()).start();

CommonData data1 = new CommonDataImpl("Key1");
new Thread(() -> data1.addData()).start();

CommonData data2 = new CommonDataImpl("Key1");
new Thread(() -> data2.addData()).start();

现在，当且仅当数据对象的键(传递给线程)相同时，我需要同步这些线程。

到目前为止我的思考过程:

我尝试创建一个类，为给定的 key 提供动态锁定，如下所示。

/*
* Singleton class
*/
public class DataAccessKeyToLockProvider {
    private volatile Map<String, ReentrantLock> accessKeyToLockHolder = new ConcurrentHashMap<>();

    private DataAccessKeyToLockProvider() {
    }

    public ReentrantLock getLock(String key) {
        return accessKeyToLockHolder.putIfAbsent(key, new ReentrantLock());
    }

    public void removeLock(BSSKey key) {
        ReentrantLock removedLock = accessKeyToLockHolder.remove(key);
    }
}

因此每个线程都会调用此类并获取锁并使用它，并在处理完成后将其删除。但这可能会导致第二个线程可以获得第一个线程插入的锁对象并等待第一个线程释放锁。一旦第一个线程删除了锁，现在第三个线程将完全获得不同的锁，因此第二个线程和第三个线程不再同步。

类似这样的事情:

   new Thread(() -> {
       ReentrantLock lock = DataAccessKeyToLockProvider.get(data.getDataAccessKey());
       lock.lock();
       data.addData();
       lock.unlock();
       DataAccessKeyToLockProvider.remove(data.getDataAccessKey());
    ).start();

如果您需要任何其他详细信息来帮助我解决问题，请告诉我

P.S:从锁提供程序中删除 key 是强制性的，因为我将处理数百万个 key (不一定是字符串)，所以我不希望锁提供程序耗尽我的内存

受到@rzwitserloot提供的解决方案的启发，我尝试放置一些通用代码，等待另一个线程完成其处理，然后再授予对下一个线程的访问权限。

public class GenericKeyToLockProvider<K> {
    private volatile Map<K, ReentrantLock> keyToLockHolder = new ConcurrentHashMap<>();

    public synchronized ReentrantLock getLock(K key) {
        ReentrantLock existingLock = keyToLockHolder.get(key);
        try {
            if (existingLock != null && existingLock.isLocked()) {
                existingLock.lock(); // Waits for the thread that acquired the lock previously to release it
            }
            return keyToLockHolder.put(key, new ReentrantLock()); // Override with the new lock
        } finally {
            if (existingLock != null) {
                existingLock.unlock();
            }
        }
    }
}

但看起来最后一个线程创建的条目不会被删除。无论如何要解决这个问题吗？

Answer 1

首先，澄清一下:您或者使用 ReentrantLock ，或者您使用 synchronized 。您不会在 ReentrantLock 实例上进行同步(您可以在任何您想要的对象上进行同步) – 或者，如果您想采用锁定路线，您可以调用 lock锁定对象上的 lock 方法，使用 try/finally 保护始终确保您调用 unlock稍后(并且根本不要使用 synchronized)。

synchronized是低级API。 Lock ，以及 java.util.concurrent 中的所有其他类包的级别更高，并提供更多的抽象。一般来说，时不时地仔细阅读 j.u.c 包中所有类的 javadoc 是一个好主意，里面有非常有用的东西。

关键问题是删除对锁对象的所有引用(从而确保它可以被垃圾收集)，但直到您确定有零个 Activity 线程锁定它。您当前的方法不知道有多少类正在等待。这需要解决。一旦您返回 Lock 对象的实例，它就“脱离了您的控制”，并且无法跟踪调用者是否会调用 lock在上面。因此，你不能这样做。相反，调用 lock 作为工作的一部分； getLock方法实际上应该将锁定作为操作的一部分。这样，您就可以控制流程。不过，让我们先退后一步:

你说你将拥有数百万把 key 。好的;但你不太可能拥有数百万个线程。毕竟线程需要栈，甚至使用-Xss参数将堆栈大小减少到最小 128k 左右，一百万个线程意味着您仅使用 128GB RAM 用于堆栈；似乎不太可能。

因此，虽然您可能拥有数百万把 key ，但“锁定” key 的数量要少得多。让我们重点关注这些。

你可以做一个 ConcurrentHashMap它将您的字符串键映射到锁定对象。然后:

获取锁:

创建一个新的锁对象(字面意思: Object o = new Object(); - 我们将使用 synchronized )并使用 putIfAbsent 将其添加到 map 中。如果您成功创建了键/值对(使用 == 将返回的对象与您创建的对象进行比较；如果它们相同，则您就是添加它的人)，您就明白了，开始运行代码。完成后，获取对象上的同步锁，发送通知，释放并删除:

public void doWithLocking(String key, Runnable op) {
    Object locker = new Object();
    Object o = concurrentMap.putIfAbsent(key, locker);
    if (o == locker) {
        op.run();
        synchronized (locker) {
            locker.notifyAll(); // wake up everybody waiting.
            concurrentMap.remove(key); // this has to be inside!
        }
    } else {
        ...
    }
}

要等到锁可用，首先获取locker对象上的锁，然后检查concurrentMap是否仍然包含它。如果没有，您现在可以重试此操作。如果它仍然存在，那么我们现在等待通知。无论如何，我们总是从头开始重试。因此:

public void performWithLocking(String key, Runnable op) throws InterruptedException {
    while (true) {
        Object locker = new Object();
        Object o = concurrentMap.putIfAbsent(key, locker);
        if (o == locker) {
            try {
                op.run();
            } finally {
                // We want to lock even if the operation throws!
                synchronized (locker) {
                    locker.notifyAll(); // wake up everybody waiting.
                    concurrentMap.remove(key); // this has to be inside!
                }
            }
            return;
        } else {
            synchronized (o) {
                if (concurrentMap.containsKey(key)) o.wait();
                }
            }
        }
    }
}

您可以使用串联的“锁定”和“解锁”方法，而不是传递要与锁定 key 一起执行的操作，但现在您面临编写忘记调用解锁的代码的风险。这就是为什么我不建议这样做!

您可以使用以下方式调用它:

keyedLockSupportThingie.doWithLocking("mykey", () -> {
    System.out.println("Hello, from safety!");
});