问

可观察对象的同步机制

霸气的gmail 发布于 2023-02-13 10:01

final

让我们假设我们必须同步对共享资源的读/写访问.多个线程将在读取和写入时访问该资源(大多数时间用于读取,有时用于写入).让我们假设每次写入总是触发读操作(对象是可观察的).

对于这个例子,我会想象一个这样的类(原谅语法和风格,它仅用于说明目的):

class Container {
    public ObservableCollection Operands;
    public ObservableCollection Results;
}

我很想将a ReadWriterLockSlim用于此目的而且我把它放在Container水平上(想象对象不那么简单,一个读/写操作可能涉及多个对象):

public ReadWriterLockSlim Lock;

的实施Operand和Result对这个例子没有意义.现在让我们想象一些观察Operands并将产生结果的代码Results:

void AddNewOperand(Operand operand) {
    try {
        _container.Lock.EnterWriteLock();
        _container.Operands.Add(operand);
    }
    finally {
        _container.ExitReadLock();
    }
}

我们的hypotetical观察者会做类似的事情,但是要使用一个新元素,它将锁定EnterReadLock()以获取操作数然后EnterWriteLock()添加结果(让我省略代码).这会因为递归而产生异常但是如果我设置LockRecursionPolicy.SupportsRecursion那么我只是打开我的代码到死锁(来自MSDN):

默认情况下,使用LockRecursionPolicy.NoRecursion标志创建ReaderWriterLockSlim的新实例,并且不允许递归.建议对所有新开发使用此默认策略,因为递归会引入不必要的复杂性并使您的代码更容易出现死锁.

为清楚起见,我重复相关部分:

递归[...]使您的代码更容易出现死锁.

如果我没有错,LockRecursionPolicy.SupportsRecursion如果来自同一个线程,我问一个,比方说,读取锁定,然后其他人要求写入锁定然后我将有一个死锁然后MSDN说的是有道理的.此外,递归也会以可测量的方式降低性能(如果我使用的话,它不是我想要的,ReadWriterLockSlim而不是ReadWriterLock或者Monitor).

问题(S)

最后我的问题是(请注意我不是在寻找关于通用同步机制的讨论,我知道这个生成器/ observable/observer场景有什么问题):

在这种情况下有什么好处？为了避免ReadWriterLockSlim赞成Monitor(即使在现实世界中,代码读取将远远超过写入)？

放弃这种粗略的同步？这甚至可以产生更好的性能,但它会使代码更加复杂(当然不是在这个例子中,而是在现实世界中).

我应该只是通知(来自观察收集)异步吗？

还有什么我看不到的东西？

我知道没有最好的同步机制,所以我们使用的工具必须是正确的,但是我想知道是否有一些最佳实践,或者我只是忽略线程和观察者之间非常重要的东西(想象一下使用Microsoft Reactive Extensions但问题是一般的,不依赖于该框架).

可能的解决方案？

我想尝试的是使事件(某种程度上)推迟:

第一个解决方案
每个更改都不会触发任何CollectionChanged事件,它会保留在队列中.当提供者(推送数据的对象)完成时,它将手动强制刷新队列(按顺序引发每个事件).这可以在另一个线程中完成,甚至可以在调用者线程中完成(但在锁定之外).

它可能有效,但它会使一切变得不那么"自动"(每个更改通知必须由生产者本身手动触发,更多代码编写,更多的错误).

第二种解决方案
另一种解决方案可能是提供对可观察集合的锁定的引用.如果我包裹ReadWriterLockSlim在一个自定义对象(有用到它藏在一个易于使用的IDisposable对象),我可以添加一个ManualResetEvent通知,所有的锁已经这样集合中被释放本身会(又在同一个线程在另一个线程)上升事件.

第三种解决方案
另一个想法可能是使事件异步.如果事件处理程序需要锁定,那么它将被停止以等待它的时间范围.为此,我担心可能使用的大线程数量(特别是如果来自线程池).

老实说,我不知道任何这些适用于现实世界的应用(个人 - 从用户的角度来看 - 我更喜欢第二个,但它意味着自定义集合的一切,这让收藏意识到线程的,我会避开它,如果可能).我不想让代码比必要的更复杂.

1 个回答

这听起来像是多线程的泡菜.在这种事件链模式中使用递归是非常具有挑战性的,同时仍然避免死锁.您可能想要考虑完全围绕问题进行设计.

例如,您可以为事件的引发异步添加操作数:

private readonly BlockingCollection<Operand> _additions
    = new BlockingCollection<Operand>();

public void AddNewOperand(Operand operand)
{
    _additions.Add(operand);
}

然后在后台线程中进行实际添加:

private void ProcessAdditions()
{
    foreach(var operand in _additions.GetConsumingEnumerable())
    {
        _container.Lock.EnterWriteLock();
        _container.Operands.Add(operand);
        _container.Lock.ExitWriteLock();
    }
}

public void Initialize()
{
    var pump = new Thread(ProcessAdditions)
    {
        Name = "Operand Additions Pump"
    };
    pump.Start();
}

这种分离牺牲了一些一致性 - 在add方法之后运行的代码实际上不知道添加实际发生的时间,也许这对您的代码来说是一个问题.如果是这样,可以重写这个以订阅观察并Task在添加完成时使用a 来发信号:

public Task AddNewOperandAsync(Operand operand)
{
    var tcs = new TaskCompletionSource<byte>();

    // Compose an event handler for the completion of this task
    NotifyCollectionChangedEventHandler onChanged = null;
    onChanged = (sender, e) =>
    {
        // Is this the event for the operand we have added?
        if (e.NewItems.Contains(operand))
        {
            // Complete the task.
            tcs.SetCompleted(0);

            // Remove the event-handler.
            _container.Operands.CollectionChanged -= onChanged;
        }
    }

    // Hook in the handler.
    _container.Operands.CollectionChanged += onChanged;

    // Perform the addition.
    _additions.Add(operand);

    // Return the task to be awaited.
    return tcs.Task;
}

事件处理程序逻辑在后台线程上引发添加消息,因此不可能阻塞前台线程.如果等待窗口的消息泵上的添加,同步上下文足够智能,也可以在消息泵线程上安排继续.

无论您是否沿着这Task条路走下去,这种策略意味着您可以安全地从可观察事件中添加更多操作数,而无需重新输入任何锁定.

2023-02-13 10:03 回答

牛奶香橙

撰写答案

今天，你开发时遇到什么问题呢？

立即提问

热门标签