使用信号量将特定线程放入队列答案

【问题标题】：Using semaphore to put particular threads in queue使用信号量将特定线程放入队列
【发布时间】：2015-08-24 08:39:08
【问题描述】：

我面临一个问题，我必须处理许多试图在数据库中写入字符串列表的线程。为了清楚起见，让我们考虑 3 个线程正在访问 DB 的连接对象。所有线程都将具有字符串。让我们坐一下，线程 A 有一个名为“First”的字符串，线程 B 有“第二”，线程 C 有“第一”。假设下面是我想要保护的代码的关键部分，我将很快解释。

writeToDataBase(String stringFromEachThread);

线程 A 和线程 B 可以同时执行此方法，因为它们内部有不同的字符串值。当线程 A 从线程 C 执行时，我必须保护上面的代码，因为线程 C 也具有与线程 A 相同的字符串值。我需要让线程 C 等到线程 A 完成执行。所以我遇到了将线程放入队列的 java.util.Concurrent.Semaphore 类。但在我的情况下，只有线程 C 必须放入队列而不是线程 B。因为线程 B 具有唯一的字符串值，所以它可以与线程 A 并行执行。信号量在线程 A 执行时阻塞线程 C，但它也阻止线程 B。下面是代码 sn-p，我在其中使用信号量来实现这一点，到目前为止这还没有帮助。我已经解决这个问题一个多星期了。任何克服这个问题的建议都非常感谢。

Semaphore sm = new Semaphore(1, true);

... ...

public void lock(String str) throws InterruptedException {
        if(!strAlreadExists()){
            return;
        }else{
            sm.acquire();
        }
}

【问题讨论】：

在我决定绝对需要信号量之前，我会考虑使用synchronized。这是关于“并发”、您拥有的选项以及这些选项之间的权衡的好教程：vogella.com/tutorials/JavaConcurrency/article.html
那是多少个字符串？此外，即使您同时执行它们，数据库也会序列化访问。那何必担心呢？
“数据库将序列化访问”--并非总是如此，这取决于隔离模式。
在您的代码方法中，strAlreadExists() 很可能不是原子的，if--else 也不是原子的，因此整个 lock() 方法不同步，因此信号量无济于事。看看synchronized 或java.util.concurrent.Lock
@SashaSalauyou 他说的不是serializable 隔离模式。如果所有的写入都进入同一个表，那么无论隔离模式如何，它们都不能同时可靠地写入。

标签： java multithreading semaphore

【解决方案1】：

您似乎要求的是为单独的字符串设置单独的锁。您的代码 sn-p 似乎表明您认为您可以使用一个锁（无论是信号量还是其他锁定机制），但不知何故不能在不相关的线程之间共享它。那是行不通的。

为单独的字符串实现单独锁定的一种方法是使用从字符串到锁定的映射。这是一个带有简单锁而不是信号量的插图：

声明一个私有映射字段：

private final ConcurrentMap<String, Object> lockMap = new ConcurrentMap<>();

为给定的字符串创建一个检索特定锁的方法：

private Object getLock( String str ) {

    Object candidateLock = new Object();
    Object returnedLock = lockMap.putIfAbsent( str, candidateLock );
    return returnedLock == null ? candidateLock : returnedLock;

}

对于任何字符串，这将首先创建一个候选锁，并尝试将其放入并发映射中。

当任何线程尝试将候选者放入映射时，映射中的该字符串已经存在锁，或者候选锁成为该字符串的新锁。使用putIfAbsent(...) 可确保带有相同字符串的所有线程都获得相同的锁对象。如果地图中有锁，候选人将被丢弃以进行垃圾收集。

现在，要访问您的关键代码，您可以：

Object lock = getLock(str);

synchronized ( lock ) {
    writeToDataBase(str);
}

如果你坚持使用信号量，你可以用new Semaphore(...)代替new Object()，用acquire代替同步。但重点仍然是每个不同的字符串都有一个单独的锁/信号量。

缺点是如果你有很多字符串和很少的重复，地图可能会变得非常大。很难从该映射中删除任何内容，除非您计划一个单独的机制来锁定所有访问该映射的线程，偶尔，并清理旧字符串。如果您使用信号量，则尤其如此，因为通常操作系统允许的信号量是有限的。

【讨论】：

非常感谢。你准确地指出了我的错误。我确实试图对不相关的线程使用相同的锁。我之前提到每个线程都有一个字符串，但实际上它是一个字符串列表。因此，如果列表中的任何字符串与另一个线程列表中存在的字符串重复，我也必须将其放入队列中。我不坚持使用信号量来解决问题。有解决办法吗？
@Parasuraman 不，那是完全不同的情况。如果这是您需要的，您应该更改您的问题以反映您的实际问题，而不是一个不真实的简化问题。我建议还显示 writeToDatabase 中需要保护的内容。

【解决方案2】：

听起来像Striped Executor Service 会很合适。

这个神奇的线程池可以保证所有具有相同stripedClass的Runnables都会按照提交的顺序执行，但是不同stripedClasses的StripedRunners仍然可以独立执行。他想使用一个相对较小的线程池来为大量 Java NIO 客户端提供服务，但以这样一种方式仍然可以按顺序执行可运行对象。

您可以使用String 作为键类，这样每个使用 asme 字符串写入的数据库都会保留 iit 的顺序，而所有其他数据库可以并行运行。

【讨论】：