ReactiveUI：如何实现周期性刷新的ReactiveList

Question

我是响应式扩展世界的新手，我仍在努力学习。

我正在开发一个带有显示某些正在运行的 Windows 进程及其内存使用情况的数据网格的应用程序。每个进程的内存使用情况应该经常更新，即每 200 毫秒。

要求

• 选中复选框时
  • 数据网格应该被进程填充，并且内存使用情况会使用间隔为 200 毫秒的计时器进行更新。
  • 监视器（所有都应该在后台线程上完成）

--如果一个进程已经退出，它应该从源中移除。

--如果一个进程启动，它应该被添加到源中

-- 用于更改的文件

• 当复选框未选中时
  • 应停止所有监视器活动
  • 数据网格被清除

任何帮助将不胜感激！ 备注：

• In the past 我尝试了几种方法，例如使用 ObservableConcurrentDictionary 作为资源和定期更新资源的计时器，但是我遇到了麻烦（并发、锁定等），所以我想有一个基于 Rx/ReactiveUI 的解决方案
• 受技术限制，我只能使用 .NET Framework 4.0，Reactive-core.Net40

更新

视图模型

private ReactiveList<IProcessModel> _processes = new ReactiveList<IProcessModel>() { ChangeTrackingEnabled = true };
public ReactiveList<IProcessModel> Processes { get { return _processes; } }

public MainViewModel(IMonitorService monitorService)
{
   this.WhenAnyValue(vm => vm.ShowProcessesIsChecked).Subscribe((b) => DoShowProcesses(b));
}


private void DoShowProcesses(bool checkboxChecked)
{
    IDisposable timer;
    Processes.Clear();
    if (checkboxChecked)
    {
        //checkbox checked
        lock (Processes)
            Processes.AddRange(_monitorService.GetProcesses());
        timer = Observable.Timer(TimeSpan.FromMilliseconds(200.0))
            .Select(x =>
        {
            lock (Processes)
            {
                foreach (var process in Processes) //throws the 'Collection was modified; enumeration operation may not execute.'
                    process.UpdateMemory(); 

                return Processes.Where(p => p.ProcessObject.HasExited).ToList();
            }
        }).
        ObserveOnDispatcher()
        .Subscribe(processesExited =>
        {
            if (processesExited.Count() > 0)
            {
                lock (Processes)
                    Processes.RemoveAll(processesExited); //remove all processes that have exited
            }

        });
    }
    else
    {
        if (timer != null)
            timer.Dispose();
    }
}

我开了一个new thread

原创

视图模型

public class MainViewModel : ReactiveObject
{
    public ReactiveList<IProcessModel> Processes { get; private set; }
    IMonitorService _monitorService;

    public MainViewModel(IMonitorService monitorService)
    {
        _monitorService = monitorService;

        Processes = new ReactiveList<IProcessModel>() { ChangeTrackingEnabled = true };
        this.WhenAnyValue(vm => vm.ShowProcessesIsChecked)
            .Where(value => value == true) //checkbox checked
            .ObserveOn(Scheduler.Default) //raise notifications on thread-pool thread to keep UI responsive
            .Select((isChecked) =>
            {
                return monitorService.GetProcesses();
            })
            .ObserveOn(SynchronizationContext.Current)
            .Subscribe(processes => {
                Processes.AddRange(processes); }
            );
        //start the MonitorService with MonitorService.Start(Processes)
        //start a timer with an interval of 200ms --> at interval
        //- do UpdateMemory() foreach IProcessModel in Processes
        //- if ProcessObject.HasExited --> remove it from the collection source
        ;
        this.WhenAnyValue(vm => vm.ShowProcessesIsChecked)
            .Where(value => value == false) //checkbox unchecked
            .Subscribe((isChecked) =>
            {
                monitorService.Stop(); //this stops monitoring for starting processes and clears the Processes
            });
    }

    private bool _showProcessesIsChecked;

    public bool ShowProcessesIsChecked
    {
        get { return _showProcessesIsChecked; }
        set { this.RaiseAndSetIfChanged(ref _showProcessesIsChecked, value); }
    }
}

型号

public class ProcessModel : ProcessModelBase, IProcessModel
{

    public ProcessModel(Process process)
    {
        ProcessObject = process;
    }      

    public void UpdateMemory()
    {
        try
        {
            if (!ProcessObject.HasExited)
            {
                long mem = ProcessObject.PagedMemorySize64;
                ProcessObject.Refresh();
                if (mem != ProcessObject.PagedMemorySize64)
                    OnPropertyChanged(nameof(ProcessObject));
            }
        }
        catch (Exception)
        {
            //log it
        }
    }
}

服务

public class MonitorService : IMonitorService
{
    ManagementEventWatcher managementEventWatcher;
    ReactiveList<IProcessModel> _processes;

    public List<IProcessModel> GetProcesses()
    {
        List<IProcessModel> processes = new List<IProcessModel>();

        foreach (var process in Process.GetProcesses().Where(p => p.ProcessName.Contains("chrome")))
            processes.Add(new ProcessModel(process));
        return processes;
    }

    /// <summary>
    /// Starts the manager. Monitor a starting process and changes in log file
    /// </summary>
    /// <param name="processes"></param>
    public void Start(ReactiveList<IProcessModel> processes)
    {
        _processes = processes;

        var qStart = "SELECT * FROM Win32_ProcessStartTrace WHERE ProcessName like 'chrome'";
        managementEventWatcher = new ManagementEventWatcher(new WqlEventQuery(qStart));
        managementEventWatcher.EventArrived += new EventArrivedEventHandler(OnProcessStarted);
        try
        {
            managementEventWatcher.Start();
        }
        catch (Exception)
        {
            //log it
        }
        Task.Factory.StartNew(() => MonitorLogFile());
    }


    public void Stop()
    {
        if (managementEventWatcher != null)
            managementEventWatcher.Stop();
        if (_processes != null)
            _processes.Clear();
    }

    private void MonitorLogFile()
    {
        //this code monitors a log file for changes. It is possible that the IsChecked property of a ProcessModel object is set in the Processes collection
    }


    private void OnProcessStarted(object sender, EventArrivedEventArgs e)
    {

        try
        {
            Process process = Process.GetProcessById(Convert.ToInt32(e.NewEvent.Properties["ProcessID"].Value));
            _processes.Add(new ProcessModel(process));
        }
        catch (ArgumentException)
        {
            //log it
        }
        catch (InvalidOperationException)
        {
            //log it
        }

    }
}

XAML

<CheckBox Content='Show Processes' IsChecked='{Binding ShowProcessesIsChecked}' />
<DataGrid  ItemsSource="{Binding Processes}">
    <DataGrid.Resources>
      <DataGridTemplateColumn Header='Process'
                              x:Key='dgProcessName'
                              IsReadOnly='True'
                              x:Shared='False'>
        <DataGridTemplateColumn.CellTemplate>
            <DataTemplate>
                <StackPanel Orientation='Horizontal' VerticalAlignment='Center'>
                    <CheckBox IsChecked='{Binding IsChecked, Mode=TwoWay, UpdateSourceTrigger=PropertyChanged}' HorizontalAlignment='Stretch' VerticalAlignment='Stretch'> </CheckBox>
                    <TextBlock Text='{Binding ProcessObject.ProcessName}' />
                </StackPanel>
            </DataTemplate>
        </DataGridTemplateColumn.CellTemplate>
          </DataGridTemplateColumn>
          <DataGridTextColumn Header="PID"
                              Binding="{Binding ProcessObject.Id}"
                              IsReadOnly='True'
                              x:Key='dgPID'
                              x:Shared='False' />
          <DataGridTextColumn Header="Commit Size"
                              Binding='{Binding ProcessObject.PagedMemorySize64}'
                              IsReadOnly='True'
                              x:Key='dgCommitSize'
                              x:Shared='False' />
    </DataGrid.Resources>
</DataGrid>

Answer 1

你基本上想使用这种模式：

IObservable<bool> checkBoxChecked = /* your checkbox observable here */
IObservable<long> timer = Observable.Interval(TimeSpan.FromMilliseconds(200.0));

IObservable<long> query =
    checkBoxChecked
        .Select(x => x ? timer : Observable.Never<long>().StartWith(-1L))
        .Switch();

IDisposable subscription =
    query
        .Subscribe(n =>
        {
            if (n == -1L)
            {
                // Clear UI
            }
            else
            {
                // Update UI
            }
        });

这会根据复选框的值在运行和不运行之间切换。

您需要确保您正在观察 UI 线程，但除了一些小的调整之外，这应该可以正常工作。

Answer 2

我还是想练习我的 Rx 技能，所以我继续创建了一个 WPF 项目并试了一下。我得到它的工作，所以我会分享我是怎么做的。

• 从 MonitorService 中删除进程列表。这将有助于隔离列表修改的来源，使我们的调试工作更轻松。它还将 MonitorService 的职责缩小到提供初始列表和发出更改。

• 我们已经在“反应”了，所以我们也可以使用FromEventPattern 将 EventArrived 事件变成一个可观察的事件。然后我们可以 transform 将这些触发的事件发送到 ProcessModels 并将它们推送给订阅者。

• 我将 ManagementEventWatcher 创建移至构造函数，因此我们不必在每次选中复选框时都重新创建它。现在，Start/Stop 方法现在只是 _managementEventWatcher 版本的包装器。

.

public class MonitorService
{
    ManagementEventWatcher _managementEventWatcher;

    public IObservable<ProcessModel> NewProcessObservable { get; }

    public MonitorService()
    {
        var qStart = "SELECT * FROM Win32_ProcessStartTrace where ProcessName='chrome.exe'";
        _managementEventWatcher = new ManagementEventWatcher(new WqlEventQuery(qStart));
        var eventArrivedObservable = Observable.FromEventPattern<EventArrivedEventHandler, EventArrivedEventArgs>(
            x => _managementEventWatcher.EventArrived += x,
            x => _managementEventWatcher.EventArrived -= x);

        NewProcessObservable = eventArrivedObservable
            .Select(x => GetProcessModel(x.EventArgs))
            .Where(x => x != null);
    }

    public List<ProcessModel> GetProcesses()
    {
        List<ProcessModel> processes = new List<ProcessModel>();

        foreach(var process in Process.GetProcesses().Where(p => p.ProcessName.Contains("chrome")))
            processes.Add(new ProcessModel(process));
        return processes;
    }

    public void Start()
    {
        try
        {
            _managementEventWatcher.Start();
        }
        catch(Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
    }

    public void Stop()
    {
        if(_managementEventWatcher != null)
            _managementEventWatcher.Stop();
    }

    private ProcessModel GetProcessModel(EventArrivedEventArgs e)
    {
        ProcessModel model = null;
        try
        {
            Process process = Process.GetProcessById(Convert.ToInt32(e.NewEvent.Properties["ProcessID"].Value));
            model = new ProcessModel(process);
        }
        catch(ArgumentException)
        {
            //log it
        }
        catch(InvalidOperationException)
        {
            //log it
        }

        return model;
    }
}

MainViewModel

• 为了避免'集合被修改等错误；枚举操作可能无法执行'一个简单的解决方案是iterate backward using a for loop。

• 在每次订阅调用之前，立即使用 .ObserveOnDispatcher()，以便 onNext 调用将在 UI 线程上执行。很酷的部分是，我们正在修改列表的唯一 位置现在在 subscribe 内部。因此，我们不必使用锁之类的东西。

• 我将逻辑划分为 3 个不同的订阅：开始/停止监控、更新内存使用并删除已退出的进程（使用 @Enigmativity 建议的模式）和 将新启动的进程添加到我们的响应式列表中。 希望这可以很容易地遵循逻辑。

.

public class MainViewModel : ReactiveObject
{
    public ReactiveList<ProcessModel> Processes { get; private set; }
    MonitorService _monitorService;

    public MainViewModel(MonitorService monitorService)
    {
        _monitorService = monitorService;

        Processes = new ReactiveList<ProcessModel>() { ChangeTrackingEnabled = true };
        RxApp.SupportsRangeNotifications = false;

        IObservable<bool> checkboxObservable = this.WhenAnyValue(vm => vm.ShowProcessesIsChecked);
        IObservable<long> intervalObservable = Observable.Interval(TimeSpan.FromMilliseconds(200.0));

        // Start/stop the monitoring.
        checkboxObservable
            // Skip the default unchecked state.
            .Skip(1)
            .ObserveOnDispatcher()
            .Subscribe(
                isChecked =>
                {
                    if(isChecked)
                    {
                        Processes.AddRange(monitorService.GetProcesses());
                        monitorService.Start();
                    }
                    else
                    {
                        Processes.Clear();
                        monitorService.Stop();
                    }
                });

        // Update the memory usage and remove processes that have exited.
        checkboxObservable
            .Select(isChecked => isChecked ? intervalObservable : Observable.Never<long>())
            // Switch disposes of the previous internal observable
            // (either intervalObservable or Never) and "switches" to the new one.
            .Switch()
            .ObserveOnDispatcher()
            .Subscribe(
                _ =>
                {
                    // Loop backwards in a normal for-loop to avoid the modification errors.
                    for(int i = Processes.Count - 1; i >= 0; --i)
                    {
                        if(Processes[i].ProcessObject.HasExited)
                        {
                            Processes.RemoveAt(i);
                        }
                        else
                        {
                            Processes[i].UpdateMemory();
                        }
                    }
                });

        // Add newly started processes to our reactive list.
        monitorService.NewProcessObservable
            .Where(_ => ShowProcessesIsChecked)
            .ObserveOnDispatcher()
            .Subscribe(process => Processes.Add(process));
    }

    private bool _showProcessesIsChecked;

    public bool ShowProcessesIsChecked
    {
        get { return _showProcessesIsChecked; }
        set { this.RaiseAndSetIfChanged(ref _showProcessesIsChecked, value); }
    }
}

更多信息 如果这是您应用程序中唯一的页面/窗口，那么这就足够了。如果没有，您还有更多工作要做，以避免内存泄漏。在后一种情况下，我建议谷歌搜索"ReactiveUI WhenActivated"（甚至可能将 wpf 扔在那里）。您会发现很多可以使用和学习的示例。