我有一组任务，它们分为两部分enque和deque，以确保合格的(多个(首先运行，然后按顺序运行其他合格的(类似于优先级队列(。
在enque内，将检查任务，只有qualified任务将被执行，而其他unqualified被阻止。
在deque中，完成的任务将从queue中删除，然后notifyAll被阻止的任务(实际上是所有其他线程来选择合格的(。
这是我想要实现的目标的简化演示：
class MyTaskQueue {
private static final Object THE_QUEUE_LOCK = new Object();
public static Map<String, ReentrantLock> taskGroupLock = new HashMap<>();
public static Map<String, Condition> taskGroupCondition = new HashMap<>();
public static void enque(String name, String taskId) {
synchronized (THE_QUEUE_LOCK) {
taskGroupLock.putIfAbsent(name, new ReentrantLock());
taskGroupCondition.putIfAbsent(name, taskGroupLock.get(name).newCondition());
}
synchronized (taskGroupLock.get(name)) {
while (true) {
if (isValid(taskId)) {
break; // Go!!;
} else {
try {
taskGroupCondition.get(name).wait(); // blocked if it's not allowed;
} catch (InterruptedException ignored) {
ignored.printStackTrace();
}
}
}
}
}
public static void deque(String name, String taskId) {
if (taskGroup.containsKey(name) && taskGroup.get(name).contains(taskId)) {
synchronized (THE_QUEUE_LOCK) {
taskGroup.get(name).remove(taskId);
if (taskGroup.get(name).isEmpty()) {
taskGroup.remove(name);
}
synchronized (taskGroupLock.get(name)) {
taskGroupCondition.get(name).notifyAll();
}
}
}
}
}
目前，尽管我检查了所有其他任务(至少其中大多数(都已正确阻止，但只会执行第一个任务。
但是当我检查taskGroupCondition.get(name)时，firstWaiter和lastWaiter都是null。
我在这里错过了什么？
任何帮助将不胜感激。
如果我理解正确，您问的是以下内容：
- 并行运行的线程请求您的
MyTaskQueue权限以开始运行(通过enque方法(。 enqueMyTaskQueue块的方法，直到请求运行的任务获得资格。- 每个合格的线程通过调用
deque方法向MyTaskQueue声明它已结束运行。 deque方法通知所有其他任务，以便检查哪些任务是合格的，然后开始运行。
然后我可以看到以下解决方案：
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Random;
import java.util.Set;
public class MyTaskQueue {
private final Map<String, Set<String>> runningTasks;
private String qualifiedTaskId;
public MyTaskQueue(String initialQualifiedTaskId) {
runningTasks = new HashMap<>();
qualifiedTaskId = initialQualifiedTaskId;
}
private synchronized boolean isValid(String taskId) {
return qualifiedTaskId != null && taskId != null && taskId.equals(qualifiedTaskId); //Do your qualification tests here...
}
public synchronized void setQualifiedTaskId(String qualifiedTaskId) {
this.qualifiedTaskId = qualifiedTaskId;
notifyAll(); //Now that the qualification test changed, is time to notify every blocked task.
//This way, all new qualified tasks will also be started. This "notifyAll()" operation is optional.
}
public synchronized void enque(String task, String taskId) {
while (!isValid(taskId)) { //Reentrant lock.
System.out.println("Blocking unqualified task {"" + task + "", "" + taskId + ""}...");
try { wait(); } catch (InterruptedException ie) { /*Handle the exception...*/ }
}
runningTasks.putIfAbsent(task, new HashSet<>());
runningTasks.get(task).add(taskId);
System.out.println("Starting qualified task {"" + task + "", "" + taskId + ""}...");
}
//Optional method. Might be needed for example if a Thread
//wants to check if another task is currently running...
public synchronized boolean isRunning(String task, String taskId) {
return runningTasks.containsKey(task) && runningTasks.get(task).contains(taskId);
}
public synchronized void deque(String task, String taskId) {
if (isRunning(task, taskId)) { //Reentrant lock.
//Cleanup:
runningTasks.get(task).remove(taskId);
if (runningTasks.get(task).isEmpty())
runningTasks.remove(task);
//Notify all blocked tasks:
notifyAll();
}
}
public static void main(final String[] args) {
MyTaskQueue q = new MyTaskQueue("qualified");
Random rand = new Random();
new MyThread(q, "Task1", "qualified222", 2500 + rand.nextInt(500)).start();
new MyThread(q, "Task2", "qualified222", 2500 + rand.nextInt(500)).start();
new MyThread(q, "Task3", "qualified", 2500 + rand.nextInt(500)).start();
new MyThread(q, "Task4", "qualified", 2500 + rand.nextInt(500)).start();
new MyThread(q, "Task5", "foreverBlocked", 2500 + rand.nextInt(500)).start();
try { Thread.sleep(3000); } catch (InterruptedException ie) { /*Handle the exception...*/ }
synchronized (q) {
System.out.println("Qualifying tasks of id "qualified222"...");
q.setQualifiedTaskId("qualified222"); //Reentrant lock.
}
//Execution of main method never ends, because of the forever blocked task "Task5".
//The "Task5" still runs while waiting for permission... See MyThread for details...
}
}
并遵循MyThread：
public class MyThread extends Thread {
private final String task, taskId;
private final int actionTime; //Dummy uptime to simulate.
private final MyTaskQueue q;
public MyThread(MyTaskQueue q, String task, String taskId, int actionTime) {
this.q = q;
this.task = task;
this.taskId = taskId;
this.actionTime = actionTime;
}
@Override
public void run() {
q.enque(task, taskId); //Wait for permission to run...
System.out.println("Task {"" + task + "", "" + taskId + ""} is currently running...");
//Now lets actually execute the task of the Thread:
try { Thread.sleep(actionTime); } catch (InterruptedException ie) { /*Handle the exception.*/ }
q.deque(task, taskId); //Declare Thread ended.
}
}
MyThread是执行所需实际操作的类。
为简单起见，我认为如果任务的 id 等于变量(即qualifiedTaskId(，则任务是限定的。
还有一个main方法来测试代码。
遵循示例输出(我为行编号(：
- 正在阻止非限定任务 {"任务 1"， "合格222"}...
- 正在阻止非限定任务 {"任务5"， "已阻止"}...
- 正在启动合格的任务 {"任务4"， "合格"}... 任务
- {"任务4"， "合格"} 当前正在运行...
- 正在启动合格的任务 {"任务3"， "合格"}... 任务
- {"任务 3"， "合格"} 当前正在运行...
- 正在阻止非限定任务 {"任务 2"， "限定 222"}...
- 正在阻止非限定任务 {"任务 2"， "限定 222"}...
- 正在阻止非限定任务 {"任务5"， "已阻止"}...
- 正在阻止非限定任务 {"任务 1"， "合格222"}...
- 正在阻止非限定任务 {"任务 1"， "合格222"}...
- 正在阻止非限定任务 {"任务5"， "已阻止"}...
- 正在阻止非限定任务 {"任务 2"， "限定 222"}...
- ID "qualified222"的合格任务...
- 正在启动合格的任务 {"任务 2"， "合格222"}... 任务
- {"任务 2"， "合格222"} 当前正在运行...
- 正在阻止非限定任务 {"任务5"， "已阻止"}...
- 正在启动合格的任务 {"任务 1"， "合格222"}... 任务
- {"任务 1"， "合格222"} 当前正在运行...
- 正在阻止非限定任务 {"任务5"， "已阻止"}...
- 正在阻止非限定任务 {"任务5"， "已阻止"}...
如您所见，第 1 行到第 7 行是每个线程的初始消息。
然后，调用第 8 行到第 10 行，因为符合条件的任务已结束(因此它们被重新阻止(。
然后，调用第 11 行到 13 行，因为另一个符合条件的任务结束了(因此它们被重新阻止(。
然后，在第 14 行到第 19 行中，资格测试发生更改，新的合格任务开始运行。还有另一个任务(Task5(尚未完成。
最后，由于任务 ID 等于 end"qualified222"限定任务，因此调用第 20 行到 21 行。