ReactiveUI中SourceCache订阅问题：属性变更无通知

要实现监听SourceCache中元素的属性变更并获取初始值通知，需要分两步处理：让对象支持属性变更通知，结合SourceCache的API实现完整的监听流。以下是具体方案：

第一步：修改Test类，实现属性变更通知 #

SourceCache本身不会自动感知对象内部属性变化，必须让Test类实现INotifyPropertyChanged接口，触发属性变更事件。同时注意原代码中name是字段，SourceCache的键选择器x => x.Name要求它是属性，需要修正：

public class Test : INotifyPropertyChanged
{
    private bool _feature1;
    private bool _feature2;
    private string _name;

    public bool Feature1
    {
        get => _feature1;
        set
        {
            if (_feature1 != value)
            {
                _feature1 = value;
                OnPropertyChanged(nameof(Feature1));
            }
        }
    }

    public bool Feature2
    {
        get => _feature2;
        set
        {
            if (_feature2 != value)
            {
                _feature2 = value;
                OnPropertyChanged(nameof(Feature2));
            }
        }
    }

    public string Name
    {
        get => _name;
        set
        {
            if (_name != value)
            {
                _name = value;
                OnPropertyChanged(nameof(Name));
            }
        }
    }

    public Test(string name)
    {
        _name = name;
        _feature1 = false;
        _feature2 = false;
    }

    public event PropertyChangedEventHandler PropertyChanged;

    protected virtual void OnPropertyChanged(string propertyName)
    {
        PropertyChanged?.Invoke(this, new PropertyChangedEventArgs(propertyName));
    }
}

第二步：选择监听方案 #

方案一：属性变更时自动更新SourceCache #

该方案利用SourceCache内置的通知机制，当对象属性变化时调用AddOrUpdate，让所有订阅者收到更新，适合多订阅者共享更新逻辑的场景。

修改Notifier类：

class Notifier
{
    public SourceCache<Test, string> testClassNotifier = new SourceCache<Test, string>(x => x.Name);
    private readonly Dictionary<string, IDisposable> _propertySubscriptions = new Dictionary<string, IDisposable>();

    public Notifier()
    {
        // 监听缓存元素移除，清理属性订阅避免内存泄漏
        testClassNotifier.WhenRemoved().Subscribe(obj =>
        {
            if (_propertySubscriptions.TryGetValue(obj.Name, out var subscription))
            {
                subscription.Dispose();
                _propertySubscriptions.Remove(obj.Name);
            }
        });

        Task.Run(() =>
        {
            AddOrUpdateSourceCache();
            SubscribeTestObj1();
            SubscribeTestObj2();
        }).ConfigureAwait(false);
    }

    private void AddOrUpdateSourceCache()
    {
        List<Test> testListObj = new List<Test>() { new Test("test1"), new Test("test2") };
        foreach (Test obj in testListObj)
        {
            testClassNotifier.AddOrUpdate(obj);
            // 订阅对象属性变化，自动更新缓存
            SubscribeToPropertyChanges(obj);
        }

        Task.Run(async () =>
        {
            for (int i = 0; i < 2; i++)
            {
                testListObj[i].Feature1 = true;
                await Task.Delay(4000).ConfigureAwait(false);
                // 此处会触发订阅通知
            }
        }).ConfigureAwait(false);
    }

    private void SubscribeToPropertyChanges(Test obj)
    {
        var key = obj.Name;
        if (_propertySubscriptions.ContainsKey(key))
        {
            _propertySubscriptions[key].Dispose();
        }

        // 属性变更时更新缓存，触发SourceCache通知
        _propertySubscriptions[key] = Observable.FromEventPattern<PropertyChangedEventHandler, PropertyChangedEventArgs>(
                h => obj.PropertyChanged += h,
                h => obj.PropertyChanged -= h)
            .Subscribe(_ => testClassNotifier.AddOrUpdate(obj));
    }

    private IObservable<Test> GetNotification(string name)
    {
        // Watch会自动发出初始值+每次AddOrUpdate的更新
        return testClassNotifier.Watch(name);
    }

    private void SubscribeTestObj1()
    {
        GetNotification("test1").Subscribe(obj =>
        {
            Console.WriteLine($"test1 状态更新: Feature1={obj.Feature1}, Feature2={obj.Feature2}");
        });
    }

    private void SubscribeTestObj2()
    {
        GetNotification("test2").Subscribe(obj =>
        {
            Console.WriteLine($"test2 状态更新: Feature1={obj.Feature1}, Feature2={obj.Feature2}");
        });
    }
}

方案二：在订阅流中组合缓存与属性变化 #

该方案无需修改缓存更新逻辑，仅在特定订阅流中处理属性变化，适合仅针对部分订阅者监听属性变更的场景：

修改GetNotification方法即可：

private IObservable<Test> GetNotification(string name)
{
    // 获取初始值
    var initialObj = testClassNotifier.Lookup(name);
    var initialObservable = initialObj.HasValue ? Observable.Return(initialObj.Value) : Observable.Empty<Test>();

    // 监听缓存中对象的替换/新增
    var cacheChanges = testClassNotifier.Watch(name);

    // 监听每个缓存对象的属性变化，包含初始状态
    var propertyChanges = cacheChanges
        .Select(obj => 
            Observable.FromEventPattern<PropertyChangedEventHandler, PropertyChangedEventArgs>(
                h => obj.PropertyChanged += h,
                h => obj.PropertyChanged -= h)
            .Select(_ => obj)
            .StartWith(obj))
        .Switch(); // 对象替换时自动切换到新对象的监听流

    // 组合初始值与所有变化通知
    return initialObservable.Concat(propertyChanges);
}

关键说明 #

• 两种方案都依赖INotifyPropertyChanged，这是监听.NET对象属性变化的标准方式。
• 方案一的优势是所有订阅SourceCache的地方都会收到属性变更通知；方案二更灵活，仅影响当前订阅流。
• 必须处理订阅的释放（如方案一中的_propertySubscriptions），避免内存泄漏。

内容的提问来源于stack exchange，提问作者Ritesh Kumar