C#事件去抖动

Question

我正在收听硬件事件消息,但我需要去除它以避免太多查询.

这是一个发送机器状态的硬件事件,我必须将其存储在数据库中以用于统计目的,并且有时会发生其状态经常变化(闪烁？).在这种情况下,我想只存储一个"稳定"状态,我想在将状态存储到数据库之前等待1-2秒来实现它.

这是我的代码:

private MachineClass connect()
{
    try
    {
        MachineClass rpc = new MachineClass();
        rpc.RxVARxH += eventRxVARxH;
        return rpc;
    }
    catch (Exception e1)
    {
        log.Error(e1.Message);
        return null;
    }
}

private void eventRxVARxH(MachineClass Machine)
{
    log.Debug("Event fired");
}

我将这种行为称为"去抖动":等待几次才能真正完成其工作:如果在去抖时间内再次触发相同的事件,我必须解除第一个请求并开始等待去抖时间以完成第二个事件.

管理它的最佳选择是什么？只是一次性计时器？

要解释"去抖"功能,请参阅以下关键事件的javascript实现:http: //benalman.com/code/projects/jquery-throttle-debounce/examples/debounce/

Answer 1

我已经用它来取消一些成功的事件:

public static Action<T> Debounce<T>(this Action<T> func, int milliseconds = 300)
{
    var last = 0;
    return arg =>
    {
        var current = Interlocked.Increment(ref last);
        Task.Delay(milliseconds).ContinueWith(task =>
        {
            if (current == last) func(arg);
            task.Dispose();
        });
    };
}

用法

Action<int> a = (arg) =>
{
    // This was successfully debounced...
    Console.WriteLine(arg);
};
var debouncedWrapper = a.Debounce<int>();

while (true)
{
    var rndVal = rnd.Next(400);
    Thread.Sleep(rndVal);
    debouncedWrapper(rndVal);
}

它可能不像RX中的那样强大,但它易于理解和使用.

Answer 2

这不是一个从头开始编码的简单要求,因为有几个细微差别.类似的情况是在尝试打开修改后的文件之前监视FileSystemWatcher并在大型副本之后等待安静.

创建.NET 4.5中的Reactive Extensions以准确处理这些场景.您可以轻松地使用它们来提供诸如Throttle,Buffer,Window或Sample等方法的功能.您将事件发布到主题,将其中一个窗口函数应用于它,例如,仅在X秒或Y事件没有活动时才获取通知,然后订阅通知.

Subject<MyEventData> _mySubject=new Subject<MyEventData>();
....
var eventSequenc=mySubject.Throttle(TimeSpan.FromSeconds(1))
                          .Subscribe(events=>MySubscriptionMethod(events));

仅当窗口中没有其他事件时,Throttle才会返回滑动窗口中的最后一个事件.任何事件都会重置窗口.

您可以在此处找到有关时移功能的非常好的概述

当您的代码收到事件时,您只需要使用OnNext将其发布到Subject:

_mySubject.OnNext(MyEventData);

如果您的硬件事件表面为典型的.NET事件,可以绕过主体和手动过帐Observable.FromEventPattern,如图所示在这里:

var mySequence = Observable.FromEventPattern<MyEventData>(
    h => _myDevice.MyEvent += h,
    h => _myDevice.MyEvent -= h);  
_mySequence.Throttle(TimeSpan.FromSeconds(1))
           .Subscribe(events=>MySubscriptionMethod(events));

您还可以从Tasks创建observable,将事件序列与LINQ运算符组合以请求例如:使用Zip的不同硬件事件对,使用另一个事件源来绑定Throttle/Buffer等,添加延迟等等.

Reactive Extensions作为NuGet包提供,因此将它们添加到项目中非常容易.

Stephen Cleary的书" C#Cookbook中的并发 "是Reactive Extensions的一个非常好的资源,并解释了如何使用它以及它如何适应.NET中的其他并发API,如任务,事件等.

Rx简介是一系列优秀的文章(我从中复制了样本),有几个例子.

UPDATE

使用您的具体示例,您可以执行以下操作:

IObservable<MachineClass> _myObservable;

private MachineClass connect()
{

    MachineClass rpc = new MachineClass();
   _myObservable=Observable
                 .FromEventPattern<MachineClass>(
                            h=> rpc.RxVARxH += h,
                            h=> rpc.RxVARxH -= h)
                 .Throttle(TimeSpan.FromSeconds(1));
   _myObservable.Subscribe(machine=>eventRxVARxH(machine));
    return rpc;
}

当然,这可以大大改善 - 可观察和订阅都需要在某个时候处理.此代码假定您只控制单个设备.如果你有很多设备,你可以在类中创建observable,这样每个MachineClass都会公开并处理它自己的observable.

Answer 3

最近我对一个针对旧版.NET框架(v3.5)的应用程序进行了一些维护.

我无法使用Reactive Extensions或任务并行库,但我需要一种漂亮,干净,一致的方法来去除事件.这是我想出的:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;

namespace MyApplication
{
    public class Debouncer : IDisposable
    {
        readonly TimeSpan _ts;
        readonly Action _action;
        readonly HashSet<ManualResetEvent> _resets = new HashSet<ManualResetEvent>();
        readonly object _mutex = new object();

        public Debouncer(TimeSpan timespan, Action action)
        {
            _ts = timespan;
            _action = action;
        }

        public void Invoke()
        {
            var thisReset = new ManualResetEvent(false);

            lock (_mutex)
            {
                while (_resets.Count > 0)
                {
                    var otherReset = _resets.First();
                    _resets.Remove(otherReset);
                    otherReset.Set();
                }

                _resets.Add(thisReset);
            }

            ThreadPool.QueueUserWorkItem(_ =>
            {
                try
                {
                    if (!thisReset.WaitOne(_ts))
                    {
                        _action();
                    }
                }
                finally
                {
                    lock (_mutex)
                    {
                        using (thisReset)
                            _resets.Remove(thisReset);
                    }
                }
            });
        }

        public void Dispose()
        {
            lock (_mutex)
            {
                while (_resets.Count > 0)
                {
                    var reset = _resets.First();
                    _resets.Remove(reset);
                    reset.Set();
                }
            }
        }
    }
}

以下是在具有搜索文本框的Windows窗体中使用它的示例:

public partial class Example : Form 
{
    private readonly Debouncer _searchDebouncer;

    public Example()
    {
        InitializeComponent();
        _searchDebouncer = new Debouncer(TimeSpan.FromSeconds(.75), Search);
        txtSearchText.TextChanged += txtSearchText_TextChanged;
    }

    private void txtSearchText_TextChanged(object sender, EventArgs e)
    {
        _searchDebouncer.Invoke();
    }

    private void Search()
    {
        if (InvokeRequired)
        {
            Invoke((Action)Search);
            return;
        }

        if (!string.IsNullOrEmpty(txtSearchText.Text))
        {
            // Search here
        }
    }
}

Answer 4

我遇到了这个问题。我在这里尝试了每个答案，并且由于我使用的是Xamarin通用应用程序，因此我似乎缺少了每个答案中都需要的某些内容，并且我不想添加任何其他程序包或库。我的解决方案完全按照我的预期工作，并且没有遇到任何问题。希望它能帮助到别人。

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace OrderScanner.Models
{
    class Debouncer
    {
        private List<CancellationTokenSource> StepperCancelTokens = new List<CancellationTokenSource>();
        private int MillisecondsToWait;
        private readonly object _lockThis = new object(); // Use a locking object to prevent the debouncer to trigger again while the func is still running

        public Debouncer(int millisecondsToWait = 300)
        {
            this.MillisecondsToWait = millisecondsToWait;
        }

        public void Debouce(Action func)
        {
            CancelAllStepperTokens(); // Cancel all api requests;
            var newTokenSrc = new CancellationTokenSource();
            lock (_lockThis)
            {
                StepperCancelTokens.Add(newTokenSrc);
            }
            Task.Delay(MillisecondsToWait, newTokenSrc.Token).ContinueWith(task => // Create new request
            {
                if (!newTokenSrc.IsCancellationRequested) // if it hasn't been cancelled
                {
                    CancelAllStepperTokens(); // Cancel any that remain (there shouldn't be any)
                    StepperCancelTokens = new List<CancellationTokenSource>(); // set to new list
                    lock (_lockThis)
                    {
                        func(); // run
                    }
                }
            });
        }

        private void CancelAllStepperTokens()
        {
            foreach (var token in StepperCancelTokens)
            {
                if (!token.IsCancellationRequested)
                {
                    token.Cancel();
                }
            }
        }
    }
}

就像这样...

private Debouncer StepperDeboucer = new Debouncer(1000); // one second

StepperDeboucer.Debouce(() => { WhateverMethod(args) });

对于机器每秒可以发送数百个请求的任何事情，我都不推荐这样做，但是对于用户输入，它的工作效果很好。我在android / IOS应用中的步进器上使用它，并在步骤中调用了api。

Answer 5

我需要这样的东西，但在网络应用程序中，所以我无法将其存储Action在变量中，它会在http请求之间丢失。

根据其他答案和@Collie 的想法，我创建了一个类，该类查看用于限制的唯一字符串键。

public static class Debouncer
{
    static ConcurrentDictionary<string, CancellationTokenSource> _tokens = new ConcurrentDictionary<string, CancellationTokenSource>();
    public static void Debounce(string uniqueKey, Action action, int seconds)
    {
        var token = _tokens.AddOrUpdate(uniqueKey,
            (key) => //key not found - create new
            {
                return new CancellationTokenSource();
            },
            (key, existingToken) => //key found - cancel task and recreate
            {
                existingToken.Cancel(); //cancel previous
                return new CancellationTokenSource();
            }
        );

        //schedule execution after pause
        Task.Delay(seconds * 1000, token.Token).ContinueWith(task =>
        {
            if (!task.IsCanceled)
            {
                action(); //run
                if (_tokens.TryRemove(uniqueKey, out var cts)) cts.Dispose(); //cleanup
            }
        }, token.Token);
    }
}

用法：

//throttle for 5 secs if it's already been called with this KEY
Debouncer.Debounce("Some-Unique-ID", () => SendEmails(), 5);

作为一个额外的好处，因为它基于字符串键，所以您可以使用内联lambda

Debouncer.Debounce("Some-Unique-ID", () => 
{
    //do some work here
}, 5);

Answer 6

RX可能是最简单的选择，尤其是如果您已经在应用程序中使用它。但是，如果没有的话，添加它可能会有点矫kill过正。

对于基于UI的应用程序（如WPF），我使用以下使用DispatcherTimer的类：

public class DebounceDispatcher
{
    private DispatcherTimer timer;
    private DateTime timerStarted { get; set; } = DateTime.UtcNow.AddYears(-1);

    public void Debounce(int interval, Action<object> action,
        object param = null,
        DispatcherPriority priority = DispatcherPriority.ApplicationIdle,
        Dispatcher disp = null)
    {
        // kill pending timer and pending ticks
        timer?.Stop();
        timer = null;

        if (disp == null)
            disp = Dispatcher.CurrentDispatcher;

        // timer is recreated for each event and effectively
        // resets the timeout. Action only fires after timeout has fully
        // elapsed without other events firing in between
        timer = new DispatcherTimer(TimeSpan.FromMilliseconds(interval), priority, (s, e) =>
        {
            if (timer == null)
                return;

            timer?.Stop();
            timer = null;
            action.Invoke(param);
        }, disp);

        timer.Start();
    }
}

要使用它：

private DebounceDispatcher debounceTimer = new DebounceDispatcher();

private void TextSearchText_KeyUp(object sender, KeyEventArgs e)
{
    debounceTimer.Debounce(500, parm =>
    {
        Model.AppModel.Window.ShowStatus("Searching topics...");
        Model.TopicsFilter = TextSearchText.Text;
        Model.AppModel.Window.ShowStatus();
    });
}

现在仅在键盘闲置200ms之后才处理键事件-丢弃任何先前的未决事件。

还有一个Throttle方法，它总是在给定间隔后触发事件：

    public void Throttle(int interval, Action<object> action,
        object param = null,
        DispatcherPriority priority = DispatcherPriority.ApplicationIdle,
        Dispatcher disp = null)
    {
        // kill pending timer and pending ticks
        timer?.Stop();
        timer = null;

        if (disp == null)
            disp = Dispatcher.CurrentDispatcher;

        var curTime = DateTime.UtcNow;

        // if timeout is not up yet - adjust timeout to fire 
        // with potentially new Action parameters           
        if (curTime.Subtract(timerStarted).TotalMilliseconds < interval)
            interval = (int) curTime.Subtract(timerStarted).TotalMilliseconds;

        timer = new DispatcherTimer(TimeSpan.FromMilliseconds(interval), priority, (s, e) =>
        {
            if (timer == null)
                return;

            timer?.Stop();
            timer = null;
            action.Invoke(param);
        }, disp);

        timer.Start();
        timerStarted = curTime;            
    }

Answer 7

这颗小宝石的灵感来自迈克·沃德 (Mike Wards) 恶魔般巧妙的延伸尝试。然而，这个东西清理后效果很好。

public static Action Debounce(this Action action, int milliseconds = 300)
{
    CancellationTokenSource lastCToken = null;

    return () =>
    {
        //Cancel/dispose previous
        lastCToken?.Cancel();
        try { 
            lastCToken?.Dispose(); 
        } catch {}          

        var tokenSrc = lastCToken = new CancellationTokenSource();

        Task.Delay(milliseconds).ContinueWith(task => { action(); }, tokenSrc.Token);
    };
}

注意：在这种情况下不需要处置该任务。请参阅此处的证据。

用法

Action DebounceToConsole;
int count = 0;

void Main()
{
    //Assign
    DebounceToConsole = ((Action)ToConsole).Debounce(50);

    var random = new Random();
    for (int i = 0; i < 50; i++)
    {
        DebounceToConsole();
        Thread.Sleep(random.Next(100));
    }
}

public void ToConsole()
{
    Console.WriteLine($"I ran for the {++count} time.");
}

Answer 8

Panagiotis的回答肯定是正确的,但是我想给出一个更简单的例子,因为我花了一段时间来分析如何让它工作.我的场景是用户在搜索框中输入,并且作为用户类型,我们想要进行api调用以返回搜索建议,因此我们要去除api调用,以便他们不会在每次键入字符时都创建一个.

我正在使用Xamarin.Android,但是这应该适用于任何C#场景......

private Subject<string> typingSubject = new Subject<string> ();
private IDisposable typingEventSequence;

private void Init () {
            var searchText = layoutView.FindViewById<EditText> (Resource.Id.search_text);
            searchText.TextChanged += SearchTextChanged;
            typingEventSequence = typingSubject.Throttle (TimeSpan.FromSeconds (1))
                .Subscribe (query => suggestionsAdapter.Get (query));
}

private void SearchTextChanged (object sender, TextChangedEventArgs e) {
            var searchText = layoutView.FindViewById<EditText> (Resource.Id.search_text);
            typingSubject.OnNext (searchText.Text.Trim ());
        }

public override void OnDestroy () {
            if (typingEventSequence != null)
                typingEventSequence.Dispose ();
            base.OnDestroy ();
        }

首次初始化屏幕/类时,您创建事件以收听用户输入(SearchTextChanged),然后还设置限制订阅,该订阅与"typingSubject"绑定.

接下来,在SearchTextChanged事件中,您可以调用typingSubject.OnNext并传入搜索框的文本.在去抖动期间(1秒)之后,它将调用订阅的事件(在我们的情况下,adviceAdapter.Get.)

最后,当屏幕关闭时,请务必处理订阅!