一段时间后使用最新数据引发事件

Question

在我的 WPF 应用程序中，UI 显示的数据更新得太频繁了。 我发现保持逻辑不变并通过一个额外的类来解决这个问题会很棒，该类存储最新的数据并在一些延迟后引发更新事件。

所以目标是更新 UI，假设每 50 毫秒，并显示最新数据。但如果没有新数据要显示，则 UI 不会更新。

这是我迄今为止创建的一个实现。有没有不加锁的方法？我的实现是否正确？

class Publisher<T>
{
    private readonly TimeSpan delay;
    private readonly CancellationToken cancellationToken;
    private readonly Task cancellationTask;

    private T data;

    private bool published = true;
    private readonly object publishLock = new object();

    private async void PublishMethod()
    {
        await Task.WhenAny(Task.Delay(this.delay), this.cancellationTask);
        this.cancellationToken.ThrowIfCancellationRequested();

        T dataToPublish;
        lock (this.publishLock)
        {
            this.published = true;
            dataToPublish = this.data;
        }
        this.NewDataAvailable(dataToPublish);
    }

    internal Publisher(TimeSpan delay, CancellationToken cancellationToken)
    {
        this.delay = delay;
        this.cancellationToken = cancellationToken;
        var tcs = new TaskCompletionSource<bool>();
        cancellationToken.Register(() => tcs.TrySetCanceled(), useSynchronizationContext: false);
        this.cancellationTask = tcs.Task;
    }

    internal void Publish(T data)
    {
        var runNewTask = false;

        lock (this.publishLock)
        {
            this.data = data;
            if (this.published)
            {
                this.published = false;
                runNewTask = true;
            }
        }

        if (runNewTask)
            Task.Run((Action)this.PublishMethod);
    }

    internal event Action<T> NewDataAvailable = delegate { };
}

Answer 1

我建议你不要重新发明轮子。 Microsoft 反应式框架非常容易处理这种情况。反应式框架允许您将事件转换为 linq 查询。

我假设您正在尝试调用 DownloadStringAsync，因此需要处理 DownloadStringCompleted 事件。

所以首先你必须把事件变成IObservable<>。这很简单：

var source = Observable
    .FromEventPattern<
        DownloadStringCompletedEventHandler,
        DownloadStringCompletedEventArgs>(
        h => wc.DownloadStringCompleted += h,
        h => wc.DownloadStringCompleted -= h);

这将返回IObservable<EventPattern<DownloadStringCompletedEventArgs>> 类型的对象。将其转换为IObservable<string> 可能会更好。这也很容易。

var sources2 =
    from ep in sources
    select ep.EventArgs.Result;

现在要实际获取值，但将它们限制为每 50 毫秒也很容易。

sources2
    .Sample(TimeSpan.FromMilliseconds(50))
    .Subscribe(t =>
    {
        // Do something with the text returned.
    });

就是这样。超级简单。

Answer 2

我会反过来做，即在 UI 线程上运行 UI 更新任务，并从那里请求数据。简而言之：

async Task UpdateUIAsync(CancellationToken token)
{
    while (true)
    {
        token.ThrowIfCancellationRequested();

        await Dispatcher.Yield(DispatcherPriority.Background);

        var data = await GetDataAsync(token);

        // do the UI update (or ViewModel update)
        this.TextBlock.Text = "data " + data;
    }
}

async Task<int> GetDataAsync(CancellationToken token)
{
    // simulate async data arrival
    await Task.Delay(10, token).ConfigureAwait(false);
    return new Random(Environment.TickCount).Next(1, 100);
}

这会在数据到达时尽快更新状态，但请注意await Dispatcher.Yield(DispatcherPriority.Background)。如果数据到达速度过快，它可以保持 UI 响应，方法是为状态更新迭代提供低于用户输入事件的优先级。

[更新] 我决定更进一步，展示当后台操作不断产生数据时如何处理这种情况。我们可能会使用Progress<T> 模式将更新发布到UI 线程（如here 所示）。这样做的问题是 Progress<T> 使用 SynchronizationContext.Post 异步排队回调。因此，当前显示的数据项在显示时可能还不是最新的。

为了避免这种情况，我创建了Buffer<T> 类，它本质上是单个数据项的生产者/消费者。它在消费者端公开async Task<T> GetData()。我在System.Collections.Concurrent 中找不到类似的东西，尽管它可能已经存在于某个地方（如果有人指出这一点，我会很感兴趣）。下面是一个完整的 WPF 应用：

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Threading;

namespace Wpf_21626242
{
    public partial class MainWindow : Window
    {
        public MainWindow()
        {
            InitializeComponent();

            this.Content = new TextBox();

            this.Loaded += MainWindow_Loaded;
        }

        async void MainWindow_Loaded(object sender, RoutedEventArgs e)
        {
            try
            {
                // cancel in 10s
                var cts = new CancellationTokenSource(10000);
                var token = cts.Token;
                var buffer = new Buffer<int>();

                // background worker task
                var workerTask = Task.Run(() =>
                {
                    var start = Environment.TickCount;
                    while (true)
                    {
                        token.ThrowIfCancellationRequested();
                        Thread.Sleep(50);
                        buffer.PutData(Environment.TickCount - start);
                    }
                });

                // the UI thread task
                while (true)
                {
                    // yield to keep the UI responsive
                    await Dispatcher.Yield(DispatcherPriority.Background);

                    // get the current data item
                    var result = await buffer.GetData(token);

                    // update the UI (or ViewModel)
                    ((TextBox)this.Content).Text = result.ToString();
                }
            }
            catch (Exception ex)
            {
                MessageBox.Show(ex.Message);
            }
        }

        /// <summary>Consumer/producer async buffer for single data item</summary>
        public class Buffer<T>
        {
            volatile TaskCompletionSource<T> _tcs = new TaskCompletionSource<T>();
            object _lock = new Object();  // protect _tcs

            // consumer
            public async Task<T> GetData(CancellationToken token)
            {
                Task<T> task = null;

                lock (_lock)
                    task = _tcs.Task;

                try
                {
                    // observe cancellation
                    var cancellationTcs = new TaskCompletionSource<bool>();
                    using (token.Register(() => cancellationTcs.SetCanceled(),
                        useSynchronizationContext: false))
                    {
                        await Task.WhenAny(task, cancellationTcs.Task).ConfigureAwait(false);
                    }

                    token.ThrowIfCancellationRequested();

                    // return the data item
                    return await task.ConfigureAwait(false);
                }
                finally
                {
                    // get ready for the next data item
                    lock (_lock)
                        if (_tcs.Task == task && task.IsCompleted)
                            _tcs = new TaskCompletionSource<T>();
                }
            }

            // producer
            public void PutData(T data)
            {
                TaskCompletionSource<T> tcs;
                lock (_lock)
                {
                    if (_tcs.Task.IsCompleted)
                        _tcs = new TaskCompletionSource<T>();
                    tcs = _tcs;
                }
                tcs.SetResult(data);
            }
        }

    }
}

Answer 3

假设您正在通过数据绑定更新您的 UI（正如您应该在 WPF 中那样），并且您在 .NET 4.5 上，您可以简单地在您的绑定表达式上使用 delay 属性而不是所有这些基础结构。

阅读一篇不错的综合文章here。

---编辑--- 我们的假模型类：

public class Model
{
    public async Task<int> GetDataAsync()
    {
        // Simulate work done on the web service
        await Task.Delay(1000);
        return new Random(Environment.TickCount).Next(1, 100);
    }
}

我们的视图模型，可以根据需要多次更新（始终在 UI 线程上）：

public class ViewModel : INotifyPropertyChanged
{
    public event PropertyChangedEventHandler PropertyChanged = delegate { };

    private readonly Model _model = new Model();
    private int _data;

    public int Data
    {
        get { return _data; }
        set
        {
            // NotifyPropertyChanged boilerplate
            if (_data != value)
            {
                _data = value;
                PropertyChanged(this, new PropertyChangedEventArgs("Data"));
            }
        }
    }

    /// <summary>
    /// Some sort of trigger that starts querying the model; for simplicity, we assume this to come from the UI thread.
    /// If that's not the case, save the UI scheduler in the constructor, or pass it in through the constructor.
    /// </summary>
    internal void UpdateData()
    {
        _model.GetDataAsync().ContinueWith(t => Data = t.Result, TaskScheduler.FromCurrentSynchronizationContext());
    }
}

最后是我们的 UI，它只会在 50 毫秒后更新，而不管视图模型属性在此期间更改了多少次：

    <TextBlock Text="{Binding Data, Delay=50}" />