具有唯一项和线程池的线程安全FIFO队列

Question

我必须在一个系统中管理计划的文件复制。文件复制是由用户安排的，我需要限制复制期间使用的系统资源数量。没有定义每次复制可能需要的时间（即，可能计划每15分钟运行一次复制，并且在下一次运行到期时，上一次运行可能仍在运行）。

我有一个调度器，它定期检查到期的文件复制，对于每个文件复制，(1)如果它没有排队也没有运行，就将它添加到阻塞队列中；（2）否则就删除它。

private final Object scheduledReplicationsLock = new Object();
private final BlockingQueue<Replication> replicationQueue = new LinkedBlockingQueue<>();
private final Set<Long> queuedReplicationIds = new HashSet<>();
private final Set<Long> runningReplicationIds = new HashSet<>();

public boolean add(Replication replication) {

    synchronized (scheduledReplicationsLock) {
        // If the replication job is either still executing or is already queued, do not add it.
        if (queuedReplicationIds.contains(replication.id) || runningReplicationIds.contains(replication.id)) {
            return false;
        }
        replicationQueue.add(replication)
        queuedReplicationIds.add(replication.id);
    }

我还有一个线程池，等待直到队列中有复制并执行它。下面是线程池中每个线程的主要方法：

public void run() {
    while (True) {
        Replication replication = null;
        synchronized (scheduledReplicationsLock) {
            // This will block until a replication job is ready to be run or the current thread is interrupted.
            replication = replicationQueue.take();

            // Move the ID value out of the queued set and into the active set
            Long replicationId = replication.getId();
            queuedReplicationIds.remove(replicationId);
            runningReplicationIds.add(replicationId);
        }
        executeReplication(replication)
    }
} 

Answer 1

保持相同的控制流，而不是在持有互斥锁的同时阻塞BlockingQueue实例，您可以Wait等待ScheduleDreplicationsLock的通知，强制工作线程释放锁并返回到等待池。

下面是您的生产者的缩小样本：

private final List<Replication> replicationQueue = new LinkedList<>();
private final Set<Long> runningReplicationIds = new HashSet<>();

public boolean add(Replication replication) {
    synchronized (replicationQueue) {
        // If the replication job is either still executing or is already queued, do not add it.
        if (replicationQueue.contains(replication) || runningReplicationIds.contains(replication.id)) {
            return false;
        } else {
            replicationQueue.add(replication);
            replicationQueue.notifyAll();
        }
    }
}

工作线程runnable将按以下方式更新：

public void run() {
    synchronized (replicationQueue) {
        while (true) {
            if (replicationQueue.isEmpty()) {
                scheduledReplicationsLock.wait();
            }
            if (!replicationQueue.isEmpty()) {
                Replication replication = replicationQueue.poll();
                runningReplicationIds.add(replication.getId())
                executeReplication(replication);
            }
        }
    }
} 

通常，您最好使用blockingqueue来协调您的生产者和复制工作者池。

顾名思义，blockingqueue本质上是阻塞，只有当无法从队列中拉出/推入项时，调用线程才会阻塞。

同时，请注意，您必须更新正在运行的/已进入队列的状态管理，因为您将只同步blockingqueue项，删除任何约束。这将取决于上下文，这是否可以接受。

private final BlockingQueue<Replication> replicationQueue = new LinkedBlockingQueue<>();

public boolean add(Replication replication) {
    // not sure if this is the proper invariant to check as at some point the replication would be neither queued nor running while still have been processed
    if (replicationQueue.contains(replication)) {
        return false;
    }
    // use `put` instead of `add` as this will block waiting for free space
    replicationQueue.put(replication);
    return true;
}

public void run() {
    while (true) {
        Replication replication = replicationQueue.take();
        executeReplication(replication);
    }
}