用于线程控制的 C# ConcurrentDictionary 与 ManualResetEvent

Question

我有一个 Windows 服务，它使用带有回调的 System.Threading.Timer 来更新端点，如下所示：

UpdateEndpointTimer = new Timer(
                new TimerCallback(UpdateSBEndpoints),
                Endpoint.Statuses.Online,
                EndpointUpdateFrequency,
                EndpointUpdateFrequency);

我的更新方法大致如下：

private void UpdateSBEndpoints(object state)
{
    ...
    using (var context = new TestHarnessContext())
    {
        var endpoints = context.Endpoints.Where(p =>
            p.Binding == Endpoint.Bindings.ServiceBus
            && p.State == Endpoint.States.Enabled
            && p.Status != status).ToList();

         foreach (var endpoint in endpoints)
         {
             //Do stuff here
         }
         ...
}

现在，由于计时器使用 ThreadPool 中的线程来触发回调，我需要采取措施来控制线程。当多个线程可以在第一个线程完成工作之前从数据库中获取相同的端点时，就会出现一个特定的问题，这会导致在 foreach 循环中完成重复的工作。

我知道该问题的两种可能的解决方案，我想知道哪一种更好，更可取。解决方案是ConcurrentDictionary 和ManualResetEvent。

在第一种情况下，我会将它放在我的 foreach 循环中，以确保一次只有一个线程可以在给定端点上运行：

if (EndpointsInAction.TryAdd(endpoint.Id, endpoint.Id) == false)
    // If we get here, another thread has started work with this endpoint.
    return;
...
//Do stuff with the endpoint, once done, remove its Id from the dictionary
...
int id;
EndpointsInAction.TryRemove(endpoint.Id, out id);

在第二种情况下，我会像这样控制线程：

protected ManualResetEvent PubIsBeingCreated { get; set; }
protected ManualResetEvent SubIsBeingCreated { get; set; }
...
this.PubIsBeingCreated = new ManualResetEvent(true);
this.SubIsBeingCreated = new ManualResetEvent(true);
...
foreach (var endpoint in endpoints)
{
   if (!this.PubIsBeingCreated.WaitOne(0))
// If we get here, another thread has started work with this endpoint.
        return;

   try
   {
       // block other threads (Timer Events)
       PubIsBeingCreated.Reset();
       // Do stuff
   }
   ...
   finally
   {
       // Restore access for other threads
       PubIsBeingCreated.Set();
   }
}

现在这两种方法似乎都有效我想知道哪种方法更可取（更有效？）。我倾向于使用ConcurrentDictionary，因为它允许对线程进行更精细的过滤，即不允许两个线程与特定端点一起工作，而不允许两个线程与特定端点type一起工作（pubs和 ManualResetEvents 中的子项）。可能有另一种解决方案优于我的解决方案，因此任何信息将不胜感激。

Answer 1

在任何情况下，您都不应为此目的使用 ManualResetEvent。使用提供更高级别抽象的对象总是更好，但即使您想在低级别控制它，使用 .NET 的 Monitor 类（通过 lock 语句，以及 Monitor.Wait() 和 Monitor.Pulse() 方法也会比使用ManualResetEvent 更好。

您似乎有一个相当标准的生产者/消费者场景。在这种情况下，在我看来，与ConcurrentDictionary 相比，使用ConcurrentQueue 会更好。您的生产者会将事物排入队列，而您的消费线程会将它们出列以进行处理。

Answer 2

我会做不同的事；我会使用 ConcurrentBag 并让一个线程从数据库加载它，多个线程通过 TryTake 监听它。加载包的单线程将解决您的问题，他们可能会从数据库中获取相同的项目。下面将在控制台应用程序中进行说明：

class Program
{
    private static ConcurrentBag<int> _bag;

    public static void Main()
    {
        // Construct and populate the ConcurrentBag
        _bag = new ConcurrentBag<int>();
        var bagPopulateTask = new Task(PopulateBag);
        bagPopulateTask.Start();

        var bagEmptyTask1 = new Task(EmptyBag);
        var bagEmptyTask2 = new Task(EmptyBag);
        var bagEmptyTask3 = new Task(EmptyBag);

        bagEmptyTask1.Start();
        bagEmptyTask2.Start();
        bagEmptyTask3.Start();

        Console.ReadKey();
    }

    private static void EmptyBag()
    {
        int i;
        while (true)
        {
            if (_bag.TryTake(out i))
            {
                Console.WriteLine("Thread {0} took {1}", Thread.CurrentThread.ManagedThreadId, i);
            }
            else
            {
                Console.WriteLine("Thread {0} took nothing", Thread.CurrentThread.ManagedThreadId);
            }
            Thread.Sleep(50);
        }
    }

    private static void PopulateBag()
    {
        var i = 0;

        while (i < 1000)
        {
            Console.WriteLine("Thread {0} added {1}", Thread.CurrentThread.ManagedThreadId, i);
            _bag.Add(i);
            i++;
            Thread.Sleep(i);
        }
    }
}

请注意，如果顺序很重要，您可以使用 ConcurrentQueue 或 ConcurrentStack 进行类似操作。