实现线程安全的并行处理

Question

我正在尝试以支持多线程和并发的方式转换现有进程，以使解决方案更加健壮和可靠。

以紧急警报系统为例。当工作人员签到时，将使用他们的信息创建一个新的收件人对象，并将其添加到收件人集合中。相反，当他们下班时，对象会被移除。在后台，当警报发生时，警报引擎将遍历相同的收件人列表 (foreach)，对每个对象调用 SendAlert(...)。

这是我的一些要求：

• 如果正在进行警报，则不应阻止添加收件人。
• 如果正在进行警报，则不应阻止删除收件人。
• 添加或删除收件人不应影响收件人列表 正在进行的警报使用的收件人。

我一直在研究 Task 和 Parallel 类以及 BlockingCollection 和 ConcurrentQueue 类，但不清楚最好的方法是什么。

是否像使用 BlockingCollection 一样简单？在阅读了大量文档后，我仍然不确定如果在枚举集合时调用 Add 会发生什么。

更新

一位同事向我推荐了以下文章，该文章描述了 ConcurrentBag 类以及每个操作的行为方式：

http://www.codethinked.com/net-40-and-system_collections_concurrent_concurrentbag

根据作者的解释，这个集合似乎（几乎）可以达到我的目的。我可以做到以下几点：

1. 创建一个新集合

   var recipients = new ConcurrentBag();

2. 当工作人员打卡时，创建一个新的收件人并将其添加到集合中：

   recipients.Add(new Recipient());

3. 当警报发生时，警报引擎可以在那时遍历集合，因为 GetEnumerator 使用集合项的快照。

   foreach（收件人中的 var 收件人） 收件人.SendAlert(...);

4. 当工作人员下班时，将收件人从集合中删除：

   ???

ConcurrentBag 不提供删除特定项目的方法。据我所知，没有一个并发课程。我错过了什么吗？除此之外，ConcurrentBag 可以满足我的一切需求。

Answer 1

ConcurrentBag<T> 绝对应该是你在这种情况下使用的最好的类。枚举的工作方式与您朋友所描述的完全一样，因此它应该适用于您布置的场景。但是，知道您必须从该集合中删除特定项目，唯一适合您的类型是 ConcurrentDictionary<K, V>。所有其他类型仅提供TryTake 方法，在ConcurrentBag<T> 的情况下，该方法是不确定的，或者在ConcurrentQueue<T> 或ConcurrentStack<T> 的情况下仅排序。

对于广播，您只需：

ConcurrentDictionary<string, Recipient> myConcurrentDictionary = ...;

...

foreach(Recipient recipient in myConcurrentDictionary.Values)
{
    ...
}

枚举器在那个瞬间再次成为字典的快照。

Answer 2

我今天早上上班，收到一封来自朋友的电子邮件，它给了我以下两个答案：

1 - 关于 Concurrent 命名空间中的集合是如何工作的，它们中的大多数都被设计为允许在不阻塞的情况下对集合进行加减运算，并且即使在枚举集合项的过程中也是线程安全的。

对于“常规”集合，获取枚举器（通过 GetEnumerator）会设置一个“版本”值，该值会被任何影响集合项的操作（例如添加、删除或清除）更改。 IEnumerator 实现会将创建时的版本集与集合的当前版本进行比较。如果不同，则会引发异常并停止枚举。

并发集合是使用段设计的，这使得支持多线程变得非常容易。但是，在枚举的情况下，它们实际上是在调用 GetEnumerator 时创建集合的快照副本，并且枚举器针对该副本工作。这允许对集合进行更改而不会对枚举器产生不利影响。当然，这意味着枚举对这些更改一无所知，但听起来您的用例允许这样做。

2 - 就您描述的具体场景而言，我认为不需要并发集合。您可以使用 ReaderWriterLock 包装标准集合，并在需要枚举时应用与 Concurrent 集合相同的逻辑。

这是我的建议：

public class RecipientCollection
{
    private Collection<Recipient> _recipients = new Collection<Recipient>();
    private ReaderWriterLock _lock = new ReaderWriterLock();

    public void Add(Recipient r)
    {
        _lock.AcquireWriterLock(Timeout.Infinite);

        try
        {
            _recipients.Add(r);
        }
        finally
        {
            _lock.ReleaseWriterLock();
        }
    }

    public void Remove(Recipient r)
    {
        _lock.AcquireWriterLock(Timeout.Infinite);

        try
        {
            _recipients.Remove(r);
        }
        finally
        {
            _lock.ReleaseWriterLock();
        }
    }

    public IEnumerable<Recipient> ToEnumerable()
    {
        _lock.AcquireReaderLock(Timeout.Infinite);

        try
        {
            var list = _recipients.ToArray();

            return list;
        }
        finally
        {
            _lock.ReleaseReaderLock();
        }
    }
}

ReaderWriterLock 确保只有在另一个更改集合内容的操作正在进行时才会阻止操作。一旦该操作完成，锁就会被释放，并且可以继续下一个操作。

您的警报引擎将使用 ToEnumerable() 方法获取当时集合的快照副本并枚举该副本。

根据发送警报和对集合进行更改的频率，这可能是一个问题，但您可能仍然可以实现某种类型的版本属性，当添加或删除项目和警报时会更改该属性引擎可以检查这个属性，看它是否需要再次调用 ToEnumerable() 来获取最新版本。或者通过在 RecipientCollection 类中缓存数组并在添加或删除项目时使缓存无效来封装它。

HTH

Answer 3

这样的实现不仅仅是并行处理方面，耐用性可能是其中最重要的。您是否考虑过使用现有的 PubSub 技术（例如...Azure Topics 或 NServiceBus）来构建它？

Answer 4

您的要求非常适合在 C# 中触发标准 .NET 事件的方式。我不知道 VB 语法是否被编译为类似的代码。标准模式看起来像：

public event EventHandler Triggered;
protected void OnTriggered()
{
    //capture the list so that you don't see changes while the
    //event is being dispatched.
    EventHandler h = Triggered;
    if (h != null)
        h(this, EventArgs.Empty);
}

或者，您可以使用不可变列表类来存储收件人。然后在发送警报时，它将首先获取当前列表并将其用作“快照”，在您发送警报时无法通过添加和删除来修改。例如：

class Alerter
{
    private ImmutableList<Recipient> recipients;

    public void Add(Recipient recipient)
    {
        recipients = recipients.Add(recipient);
    }

    public void Remove(Recipient recipient)
    {
        recipients = recipients.Remove(recipient);
    }

    public void SendAlert()
    {
        //make a local reference to the current list so
        //you are not affected by any calls to Add/Remove
        var current = recipients;
        foreach (var r in current)
        {
            //send alert to r
        }
    }
}

您必须找到一个 ImmutableList 的实现，但您应该能够找到几个而不需要太多工作。在我写的 SendAlert 方法中，我可能不需要显式地创建本地来避免问题，因为 foreach 循环本身就会这样做，但我认为副本使意图更清晰。