如何处理 IDisposable 值的 Observable

Question

底漆；在我的代码库中，我经常需要使用池化的内存块。这样做是出于性能原因以减少垃圾收集（制作实时视频游戏引擎组件）。我通过将类型公开为 IDisposableValue 来处理此问题，您只能在其中访问 T Value 直到包装器被释放。您处理包装器以将值返回到池中以供重用。

我构建了处理这些包装值的数据流，以响应随时间发生的事件。这通常是 Observables/Reactive Extensions 的完美候选者，除了必须处理 wrapper 本质上是一种可变性形式，这是您在响应式时不想要的。如果一个订阅者在处理完包装器后释放了它，但第二个观察者仍在使用它，那么包装器将抛出异常。

预期目标：让每个订阅者收到一个单独的包装器，而不是原始的实际包装值。然后，只有在每个订阅者处理了他们各自的包装器（想想 RefCountDisposable）后，才会处理基础值。因此，每个订阅者都可以根据需要使用该值，并且它们表示它们是通过处置完成的。当它们全部完成时，值将被释放回池中。

唯一的问题是我不知道如何在 RX 中正确实现这一点。这是处理我的情况的适当方法吗？如果是，有任何关于如何实际实施它的指示吗？

编辑 1 - 使用 ISubject 的脏解决方案：

我尝试使用 Observable.Select/Create/Defer 的各种组合使其工作，但无法让上面的 Intended Goal 工作。相反，我不得不转向使用主题，我知道这是回避的。这是我当前的代码。

public class SharedDisposableValueSubject<T> : AbstractDisposable, ISubject<IDisposableValue<T>>
{
    private readonly Subject<SharedDisposable> subject;

    private readonly SubscriptionCounter<SharedDisposable> counter;

    private readonly IObservable<IDisposableValue<T>> observable;

    public SharedDisposableValueSubject()
    {
        this.subject = new Subject<SharedDisposable>();
        this.counter = new SubscriptionCounter<SharedDisposable>(this.subject);
        this.observable = this.counter.Source.Select(value => value.GetValue());
    }

    /// <inheritdoc />
    public void OnCompleted() => this.subject.OnCompleted();

    /// <inheritdoc />
    public void OnError(Exception error) => this.subject.OnError(error);

    /// <inheritdoc />
    public void OnNext(IDisposableValue<T> value) =>
        this.subject.OnNext(new SharedDisposable(value, this.counter.Count));

    /// <inheritdoc />
    public IDisposable Subscribe(IObserver<IDisposableValue<T>> observer) => this.observable.Subscribe(observer);

    /// <inheritdoc />
    protected override void ManagedDisposal() => this.subject.Dispose();

    private class SharedDisposable
    {
        private readonly IDisposableValue<T> value;

        private readonly AtomicInt count;

        public SharedDisposable(IDisposableValue<T> value, int count)
        {
            Contracts.Requires.That(count >= 0);

            this.value = value;
            this.count = new AtomicInt(count);

            if (count == 0)
            {
                this.value?.Dispose();
            }
        }

        public IDisposableValue<T> GetValue() => new ValuePin(this);

        private class ValuePin : AbstractDisposable, IDisposableValue<T>
        {
            private readonly SharedDisposable parent;

            public ValuePin(SharedDisposable parent)
            {
                Contracts.Requires.That(parent != null);

                this.parent = parent;
            }

            /// <inheritdoc />
            public T Value => this.parent.value != null ? this.parent.value.Value : default(T);

            /// <inheritdoc />
            protected override void ManagedDisposal()
            {
                if (this.parent.count.Decrement() == 0)
                {
                    this.parent.value?.Dispose();
                }
            }
        }
    }
}

public class SubscriptionCounter<T>
{
    private readonly AtomicInt count = new AtomicInt(0);

    public SubscriptionCounter(IObservable<T> source)
    {
        Contracts.Requires.That(source != null);

        this.Source = Observable.Create<T>(observer =>
        {
            this.count.Increment();
            return new Subscription(source.Subscribe(observer), this.count);
        });
    }

    public int Count => this.count.Read();

    public IObservable<T> Source { get; }

    private class Subscription : AbstractDisposable
    {
        private readonly IDisposable subscription;

        private readonly AtomicInt count;

        public Subscription(IDisposable subscription, AtomicInt count)
        {
            Contracts.Requires.That(subscription != null);
            Contracts.Requires.That(count != null);

            this.subscription = subscription;
            this.count = count;
        }

        /// <inheritdoc />
        protected override void ManagedDisposal()
        {
            this.subscription.Dispose();
            this.count.Decrement();
        }
    }
}

public interface IDisposableValue<out T> : IDisposable
{
    bool IsDisposed { get; }

    T Value { get; }
}

AbstractDisposable 只是一次性模式的基类实现，用于不包含非托管类型的类型。它确保 ManagedDisposal() 仅在第一次调用 Dispose() 时被调用一次。 AtomicInt 是 Int 上 Interlocked 的包装器，用于为 int 提供线程安全的原子更新。

我的测试代码显示了预计如何使用 SharedDisposableValueSubject；

public static class SharedDisposableValueSubjectTests
{
    [Fact]
    public static void NoSubcribersValueAutoDisposes()
    {
        using (var subject = new SharedDisposableValueSubject<int>())
        {
            var sourceValue = new DisposableWrapper<int>(0);
            sourceValue.IsDisposed.Should().BeFalse();

            subject.OnNext(sourceValue);
            sourceValue.IsDisposed.Should().BeTrue();

            subject.OnCompleted();
        }
    }

    [Fact]
    public static void SingleSurcriber()
    {
        using (var subject = new SharedDisposableValueSubject<int>())
        {
            var testNumber = 1;
            var sourceValue = new DisposableWrapper<int>(testNumber);
            sourceValue.IsDisposed.Should().BeFalse();

            IDisposableValue<int> retrieved = null;
            subject.Subscribe(value => retrieved = value);

            // value retrieved from sequence but not disposed yet
            subject.OnNext(sourceValue);
            retrieved.Should().NotBeNull();
            retrieved.Value.Should().Be(testNumber);
            retrieved.IsDisposed.Should().BeFalse();
            sourceValue.IsDisposed.Should().BeFalse();

            // disposing retrieved disposes the source value
            retrieved.Dispose();
            retrieved.IsDisposed.Should().BeTrue();
            sourceValue.IsDisposed.Should().BeTrue();

            subject.OnCompleted();
        }
    }

    [Fact]
    public static void ManySubcribers()
    {
        using (var subject = new SharedDisposableValueSubject<int>())
        {
            var testNumber = 1;
            var sourceValue = new DisposableWrapper<int>(testNumber);
            sourceValue.IsDisposed.Should().BeFalse();

            IDisposableValue<int> retrieved1 = null;
            subject.Subscribe(value => retrieved1 = value);
            IDisposableValue<int> retrieved2 = null;
            subject.Subscribe(value => retrieved2 = value);

            // value retrieved from sequence but not disposed yet
            subject.OnNext(sourceValue);
            retrieved1.Should().NotBeNull();
            retrieved1.Value.Should().Be(testNumber);
            retrieved1.IsDisposed.Should().BeFalse();
            retrieved2.Should().NotBeNull();
            retrieved2.Value.Should().Be(testNumber);
            retrieved2.IsDisposed.Should().BeFalse();
            sourceValue.IsDisposed.Should().BeFalse();

            // disposing only 1 retrieved value does not yet dispose the source value
            retrieved1.Dispose();
            retrieved1.IsDisposed.Should().BeTrue();
            retrieved2.IsDisposed.Should().BeFalse();
            retrieved2.Value.Should().Be(testNumber);
            sourceValue.IsDisposed.Should().BeFalse();

            // disposing both retrieved values disposes the source value
            retrieved2.Dispose();
            retrieved2.IsDisposed.Should().BeTrue();
            sourceValue.IsDisposed.Should().BeTrue();

            subject.OnCompleted();
        }
    }

    [Fact]
    public static void DisposingManyTimesStillRequiresEachSubscriberToDispose()
    {
        using (var subject = new SharedDisposableValueSubject<int>())
        {
            var testNumber = 1;
            var sourceValue = new DisposableWrapper<int>(testNumber);
            sourceValue.IsDisposed.Should().BeFalse();

            IDisposableValue<int> retrieved1 = null;
            subject.Subscribe(value => retrieved1 = value);
            IDisposableValue<int> retrieved2 = null;
            subject.Subscribe(value => retrieved2 = value);

            subject.OnNext(sourceValue);

            // disposing only 1 retrieved value does not yet dispose the source value
            // even though the retrieved value is disposed many times
            retrieved1.Dispose();
            retrieved1.Dispose();
            retrieved1.Dispose();
            retrieved1.IsDisposed.Should().BeTrue();
            retrieved2.IsDisposed.Should().BeFalse();
            sourceValue.IsDisposed.Should().BeFalse();

            // disposing both retrieved values disposes the source value
            retrieved2.Dispose();
            retrieved2.IsDisposed.Should().BeTrue();
            sourceValue.IsDisposed.Should().BeTrue();

            subject.OnCompleted();
        }
    }

    [Fact]
    public static void SingleSubcriberUnsubcribes()
    {
        using (var subject = new SharedDisposableValueSubject<int>())
        {
            var testNumber = 1;
            var sourceValue = new DisposableWrapper<int>(testNumber);
            sourceValue.IsDisposed.Should().BeFalse();

            var subscription = subject.Subscribe(value => { });
            subscription.Dispose();

            // source value auto disposes because no subscribers
            subject.OnNext(sourceValue);
            sourceValue.IsDisposed.Should().BeTrue();

            subject.OnCompleted();
        }
    }

    [Fact]
    public static void SubcriberUnsubcribes()
    {
        using (var subject = new SharedDisposableValueSubject<int>())
        {
            var testNumber = 1;
            var sourceValue = new DisposableWrapper<int>(testNumber);
            sourceValue.IsDisposed.Should().BeFalse();

            IDisposableValue<int> retrieved = null;
            subject.Subscribe(value => retrieved = value);

            var subscription = subject.Subscribe(value => { });
            subscription.Dispose();

            // value retrieved from sequence but not disposed yet
            subject.OnNext(sourceValue);
            retrieved.Should().NotBeNull();
            retrieved.Value.Should().Be(testNumber);
            retrieved.IsDisposed.Should().BeFalse();
            sourceValue.IsDisposed.Should().BeFalse();

            // disposing retrieved causes source to be disposed
            retrieved.Dispose();
            retrieved.IsDisposed.Should().BeTrue();
            sourceValue.IsDisposed.Should().BeTrue();

            subject.OnCompleted();
        }
    }

    [Fact]
    public static async Task DelayedSubcriberAsync()
    {
        using (var subject = new SharedDisposableValueSubject<int>())
        {
            var testNumber = 1;
            var sourceValue = new DisposableWrapper<int>(testNumber);
            sourceValue.IsDisposed.Should().BeFalse();

            // delay countdown event used just to ensure that the value isn't disposed until assertions checked
            var delay = new AsyncCountdownEvent(1);
            var disposed = new AsyncCountdownEvent(2);

            subject.Delay(TimeSpan.FromSeconds(1)).Subscribe(async value =>
            {
                await delay.WaitAsync().DontMarshallContext();
                value.Dispose();
                disposed.Signal(1);
            });

            subject.Subscribe(value =>
            {
                value.Dispose();
                disposed.Signal(1);
            });

            // value is not yet disposed
            subject.OnNext(sourceValue);
            sourceValue.IsDisposed.Should().BeFalse();

            // wait for value to be disposed
            delay.Signal(1);
            await disposed.WaitAsync().DontMarshallContext();
            sourceValue.IsDisposed.Should().BeTrue();

            subject.OnCompleted();
        }
    }

    [Fact]
    public static void MultipleObservedValues()
    {
        using (var subject = new SharedDisposableValueSubject<int>())
        {
            var testNumber1 = 1;
            var sourceValue1 = new DisposableWrapper<int>(testNumber1);
            sourceValue1.IsDisposed.Should().BeFalse();

            var testNumber2 = 2;
            var sourceValue2 = new DisposableWrapper<int>(testNumber2);
            sourceValue2.IsDisposed.Should().BeFalse();

            IDisposableValue<int> retrieved = null;
            subject.Subscribe(value => retrieved = value);

            // first test value
            // value retrieved from sequence but not disposed yet
            subject.OnNext(sourceValue1);
            retrieved.Should().NotBeNull();
            retrieved.Value.Should().Be(testNumber1);
            retrieved.IsDisposed.Should().BeFalse();
            sourceValue1.IsDisposed.Should().BeFalse();

            // disposing retrieved disposes the source value
            retrieved.Dispose();
            retrieved.IsDisposed.Should().BeTrue();
            sourceValue1.IsDisposed.Should().BeTrue();

            // second test value
            // value retrieved from sequence but not disposed yet
            subject.OnNext(sourceValue2);
            retrieved.Should().NotBeNull();
            retrieved.Value.Should().Be(testNumber2);
            retrieved.IsDisposed.Should().BeFalse();
            sourceValue2.IsDisposed.Should().BeFalse();

            // disposing retrieved disposes the source value
            retrieved.Dispose();
            retrieved.IsDisposed.Should().BeTrue();
            sourceValue2.IsDisposed.Should().BeTrue();

            subject.OnCompleted();
        }
    }
}

所有这些都通过了，但我意识到你可以用 observable 做很多事情，所以可能有一些我没有考虑过的用例可能会破坏这个实现。如果您知道任何问题，请告诉我。也可能是我试图让 Rx 做一些它本来不应该做的事情。

编辑 2 - 使用发布的解决方案：

我使用 Publish 将原始 observable 中的一次性值包装在 SharedDisposable 中，确保每个原始值仅被包装一次。然后发布的 observable 被订阅者计数，并且每个订阅者都会获得一个单独的 ValuePin，当处置该值时会减少 SharedDisposable 上的计数。当 SharedDisposable 计数达到 0 时，它会释放原始值。

我尝试不进行订阅计数，而是使用它，因此每次发出 ValuePin 时都会增加计数，但我无法找到一种方法来保证它会在允许订阅者之前为每个订阅者创建 ValuePin处置它们。这导致订阅者 1 获得他们的 pin，计数从 0 变为 1，然后在订阅者 2 获得他们的 pin 之前处理该 pin，计数从 1 变为 0 触发原始值被处理，然后订阅者 2 将收到一个别针，但现在为时已晚。

public static IObservable<IDisposableValue<T>> ShareDisposable<T>(this IObservable<IDisposableValue<T>> source)
{
    Contracts.Requires.That(source != null);

    var published = source.Select(value => new SharedDisposable<T>(value)).Publish();
    var counter = new SubscriptionCounter<SharedDisposable<T>>(published);
    published.Connect();
    return counter.CountedSource.Select(value => value.GetValue(counter.Count));
}

private class SharedDisposable<T>
{
    private const int Uninitialized = -1;

    private readonly IDisposableValue<T> value;

    private readonly AtomicInt count;

    public SharedDisposable(IDisposableValue<T> value)
    {
        this.value = value;
        this.count = new AtomicInt(Uninitialized);
    }

    public IDisposableValue<T> GetValue(int subscriberCount)
    {
        Contracts.Requires.That(subscriberCount >= 0);

        this.count.CompareExchange(subscriberCount, Uninitialized);
        return new ValuePin(this);
    }

    private class ValuePin : AbstractDisposable, IDisposableValue<T>
    {
        private readonly SharedDisposable<T> parent;

        public ValuePin(SharedDisposable<T> parent)
        {
            Contracts.Requires.That(parent != null);

            this.parent = parent;
        }

        /// <inheritdoc />
        public T Value => this.parent.value != null ? this.parent.value.Value : default(T);

        /// <inheritdoc />
        protected override void ManagedDisposal()
        {
            if (this.parent.count.Decrement() == 0)
            {
                this.parent.value?.Dispose();
            }
        }
    }
}

这当然看起来更好，因为我不必以任何方式使用主题，尽管订阅者计数感觉很脏。特别是因为我需要在第一个 ValuePin 给出之前未初始化计数。需要明确的是，我正在尝试处理由 0 共享给许多订阅者的 observable 产生的值的处置，而不是处置与 observable 本身的连接，这就是为什么我不使用 RefCount 而不是 Connect .

Answer 1

我认为您可以重新计算一次性用品。这将要求发布者启动引用计数，然后每个订阅者递增和递减计数器。您可以使用 RefCountDisposable 来执行此操作。我只会考虑对私有/内部代码执行此操作，否则您可能会让泄漏的消费者破坏您的系统。 Rx 的替代解决方案可能是查看 Disruptor 模式。