如何动态锁定线程并避免竞争条件

Question

我正在尝试动态锁定线程，但无论我尝试什么，总是会发生某种竞争条件，这似乎是由多个线程同时启动任务引起的，但我一直无法找到很好的答案。

这是一个例子：

using System.Linq;
using System.Threading;
using System.Threading.Tasks;

public class Example
{
public static readonly Object padLock = new Object();
public readonly static ConcurrentDictionary<string, object> locks = new 
ConcurrentDictionary<string, object>();
public static List<string> processes = new List<string>();

public static void Main()
{
  //Add random list of processes (just testing with one for now)
  for (var i = 0; i < 1; i++) {
     processes.Add("random" + i.ToString()); 
  }

  while (true)
  {
     foreach (var process in processes )
     {
        var currentProc = process ;
        lock (padLock)
        {
           if (!locks.ContainsKey(currentProc))
           {
              System.Threading.Tasks.Task.Factory.StartNew(() =>
              {
                 if (!locks.ContainsKey(currentProc))
                 {
                    var lockObject = locks.GetOrAdd(currentProc, new object());
                    lock (lockObject)
                    { 
                       Console.WriteLine("Currently Executing " + currentProc); 
                       Console.WriteLine("Ended Executing " + currentProc);
                       ((IDictionary)locks).Remove(currentProc);
                    }
                 }
              });
           }
        }
       // Thread.Sleep(0);
     }
  }

  Console.ReadLine();
 }
}

输出：

Started 1

Finished 1

Started 1

Finished 1

Started 1

Finished 1

但有时会得到：

Started 1

Started 1

Finished 1

这是不希望的，动态锁应该锁定它并只执行一次

Answer 1

虽然我不知道您要做什么，但您看到的行为是因为您从多个线程向 Console.WriteLine 发送垃圾邮件，它们被乱序打印。我通过将时间戳附加到 Console.WriteLine 来验证这一点：

public class Example
{
  public static readonly Object padLock = new Object();
  public readonly static ConcurrentDictionary<string, object> locks = new ConcurrentDictionary<string, object>(); 
  public static List<string> processes = new List<string>(); 

  [ThreadStatic]
  private static bool flag = false;

 public static void Main()
 {
  //Add random list of processes (just testing with one for now)
  for (var i = 0; i < 10; i++)
  {
     processes.Add(i.ToString());
  }

  while (true)
  {
     foreach (var process in processes)
     {
        var currentProc = process; 

        if (!locks.ContainsKey(currentProc))
        {
           var lockObject = locks.GetOrAdd(currentProc, new object());
           Task.Factory.StartNew(() =>
            {
               lock (lockObject)
               {
                  if (flag) throw new Exception();
                  flag = true;
                  Console.WriteLine("Currently Executing " + currentProc);
                  Thread.Sleep(0); // You can siimulate work here
                  Console.WriteLine("Ended Executing " + currentProc);
                  flag = false;
                  ((IDictionary)locks).Remove(currentProc);
               }
            });
        }
     }
   }
  }
}

运行如下：

....
Ended Executing 1 4025734
Currently Executing 1 4026419
Ended Executing 1 4028737
Currently Executing 1 4029565
Ended Executing 1 4030472
Currently Executing 1 4031643
Currently Executing 1 3659670
Ended Executing 1 4032900
Currently Executing 1 4033582
Ended Executing 1 4034318
Ended Executing 1 4032786
Currently Executing 1 4038042
Ended Executing 1 4042484
Currently Executing 1 4044967
Currently Executing 1 4007282
...

Answer 2

这是一个常见的问题

我将您的要求解读为“获取进程列表，然后使用多个线程对每个进程执行一次操作。”

就我的示例而言，假设 Foo(process) 完成了必须只执行一次的工作单元。

这是一个非常普遍的需求，有几种模式。

Parallel.ForEach

这种技术可以为循环的每次迭代使用不同的线程，这些线程将同时执行。

Parallel.ForEach(processes, process => Foo(process));

是的；就一行代码。

异步任务

如果Foo() 是异步的，这种技术是合适的。它只是为所有进程安排一个任务，然后等待它们，并让 SynchronizationContext 对其进行排序。

var tasks = processes.Select( p => Foo(process) );
await Task.WhenAll(tasks);

生产者/消费者

这使用producer-consumer pattern，这是一个线程添加到队列而另一个线程从队列中取出的传统方式。通过从队列中删除一个项目，它实际上被“锁定”了，这样其他线程就不会尝试处理它。

BlockingCollection<string>() queue = new BlockingCollection<string>();

void SetUpQueue()
{
    for (int i=0; i<100; i++) queue.Add(i.ToString());
    queue.CompleteAdding();
}

void Worker()
{
    while (queue.Count > 0 || !queue.IsAddingCompleted)
    {
        var item = queue.Take();
        Foo(item);
    }
}