c# - 确保顺序处理多个 IDisposables

Question

我有两个 IDisposables 需要按顺序处理。排序很重要，因为第一个 IDisposable 终止了依赖于将被第二个 IDisposable 终止的服务的 Rx 订阅。这是在 Windows 窗体应用程序中，其中 IObservable 的订阅需要在不同的线程上发生，但观察和处理需要在 UI 线程上发生。 (实际上，只要确保顺序，我不关心是否在 UI 线程上进行处理。)因此，在代码中我大致具有以下内容(一旦减少):

SomeService = new DisposableService();
Subscription = Foo(someService).SubscribeOn(NewThreadScheduler.Default).ObserveOn(theForm).Subscribe(...)

在许多 UI 事件中，我需要按顺序处理这两个事件(订阅，然后是 SomeService)。为此，除了 ContextDisposable 之外，我还尝试使用 Rx 的 CompositeDisposable 在同一线程上提供串行处理:

_Disposable = new CompositeDisposable(new[] {                     
    new ContextDisposable(WindowsFormsSynchronizationContext.Current, Subscription),                     
    new ContextDisposable(WindowsFormsSynchronizationContext.Current, SomeService)});

但是上面的方法不起作用。根据我的日志记录，_Disposable 和 SomeService 的 ContextDisposable 在同一线程上被调用，但 ContextDisposable 仍在发生在与正在处理的服务同时进行的不同线程上(从而导致竞争条件和 NPE)。

我使用 C# 编程才几个星期，所以我确定问题出在我对上下文和调度程序的工作方式的误解。解决这个问题的正确方法是什么？

Answer 1

除非我误解了什么，否则您可以控制哪个线程处理什么。谁订阅哪个线程并不重要。看这个例子

internal class Program
{
    private static void Main(string[] args)
    {
        ReactiveTest rx1 = null;
        ReactiveTest rx2 = null;

        var thread1 = new Thread(() => rx1 = new ReactiveTest());
        var thread2 = new Thread(() => rx2 = new ReactiveTest());

        thread1.Start();
        thread2.Start();

        Thread.Sleep(TimeSpan.FromSeconds(1));

        thread1.Join();
        thread2.Join();

        rx1.Dispose();
        rx2.Dispose();
    }
}

public class ReactiveTest : IDisposable
{
    private IDisposable _timerObservable;

    private object _lock = new object();

    public ReactiveTest()
    {
        _timerObservable = Observable.Interval(TimeSpan.FromMilliseconds(250)).Subscribe(i => 
            Console.WriteLine("[{0}] - {1}", Thread.CurrentThread.ManagedThreadId, i));
    }

    public void Dispose()
    {
        lock (_lock)
        {
            _timerObservable.Dispose();
            Console.WriteLine("[{0}] - DISPOSING", Thread.CurrentThread.ManagedThreadId);
        }
    }
}

这输出

[14] - 0
[7] - 0
[15] - 1
[7] - 1
[14] - 2
[15] - 2
[10] - DISPOSING
[10] - DISPOSING

您可以看到我们订阅了两个单独的线程，然后在第三个线程上进行了处理。我只是锁定了处理，以防你有一些线程安全需要在订阅中发生。在这个例子中，它真的没有必要。