线程顺序同步

Question

假设我有一个任务队列，每个任务都有一个锁定对象（syncObject）来控制共享资源访问，队列可以有多个任务共享相同的同步对象实例。而且我有 N 个并发线程应该使任务出队并按队列顺序处理它们，这意味着按队列顺序获取同步对象的锁定。

代码说明：

abstract class Task
{
    public readonly Object SyncObject = new Object();
}

Queue<Task> taskQueue = new Queue<Task>();
Object queueLock = new Object();

void TakeItemThreadedMethod()
{
    Task task;
    lock(queueLock) task = taskQueue.Dequeue();
    //Between this lines is my problem,
    //Other thread can dequeue next task and it may share same syncObject and
    //acquire lock on it before this thread, thought this task was first in queue
    lock(task.SyncObject)
    {
        //Do some work here
    }
}

如何开始处理按队列中的顺序共享相同 SyncObject 的任务（获取 Task.SyncObject 锁）。

Answer 1

听起来您的队列可能不应该包含单个任务 - 而是任务队列，其中每个子队列是“共享同步锁的所有任务”。

因此，您的处理器会：

• 从主队列中取出一个子队列
• 将第一个任务从子队列中取出并处理它
• 完成后将子队列放回主队列的末尾（或任何地方，实际上 - 确定您希望调度如何工作）

这将确保每个子队列一次只执行一个任务。

您可能需要从锁到子队列的映射，以便任何创建 工作都可以将其添加到正确的子队列。假设您完全需要该功能，您需要自动计算何时从地图中删除子队列（而不是将其放回主队列）。

编辑：作为对上述内容的优化，您可以将子队列本身放入您用作共享同步锁的任何内容中。它可以引用或者“下一次执行的单个任务”或“任务队列”——只是懒惰地创建队列。然后，您将同步锁（实际上不再需要用作锁）放在队列中，每个消费者只会要求它执行下一个任务。如果只有一个任务可用，则返回它（并且“下一个任务”变量设置为 null）。如果有多个任务可用，则将第一个任务出列。

当生产者添加新任务时，如果“第一个任务”变量之前为空，则将其设置为要执行的任务，或者如果没有队列，则创建队列但已经是一项任务，或者如果队列已经存在，则只是添加到队列中。这解决了不必要的队列创建的低效率。

同样，棘手的部分将是解决如何以原子方式丢弃共享资源锁 - 因为您只想在处理最后一项之后这样做，但同样您不想这样做错过了一项任务，因为您碰巧在错误的时间添加了它。它应该不会太糟糕，但同样你需要仔细考虑。

Answer 2

这个方法怎么样：

• 使用列表而不是队列
• 让每个工作线程按顺序循环通过队列，直到找到“解锁”任务

类似的东西（未经测试）：

abstract class Task
{
    public readonly Object SyncObject = new Object();
}

List<Task> taskList = new List<Task>();

void TakeItemThreadedMethod()
{
    Task task = null;
    bool found = false;

    try
    {
        // loop until found an task whose SyncObject is free
        while (!found)
        {
            lock (taskList)
            {
                for (int i = 0; i < taskList.Count; i++)
                {
                    object syncObj = taskList[i].SyncObject;
                    if (found = Monitor.TryEnter(syncObj))
                    {
                        for (int x = 0; x < taskList.Count; x++)
                        {
                            if (Object.ReferenceEquals(
                                    syncObj, taskList[x].SyncObject))
                            {
                                task = taskList[x];
                                taskList.RemoveAt(x);
                                break;
                            }
                        }
                        break;
                    }
                }
            }
        }

        // process the task...
        DoWork(task);
    }
    finally
    {
        if (found) Monitor.Exit(task.SyncObject);
    }
}

void QueueTask(Task task)
{
    lock (taskList)
    {
        taskList.Add(task);
    }
}

Answer 3

我用过QueuedLock class suggested by Matthew Brindley，稍作修改，我将Enter函数拆分为TakeTicket和Enter which blocks。

现在我可以在共享 QueueLock 中使用 TakeTicket 而不会阻塞整个队列。

修改代码：

abstract class Task
{
    public readonly QueuedLock SyncObject = new QueuedLock();
}

Queue<Task> taskQueue = new Queue<Task>();
Object queueLock = new Object();

void TakeItemThreadedMethod()
{
    Task task;
    int ticket;
    lock(queueLock) 
    {
        task = taskQueue.Dequeue();
        ticket = task.SyncObject.TakeTicket();
    }
    task.SyncObject.Enter(ticket);
    //Do some work here
    task.SyncObject.Exit();
}