19

所以我的要求是让我的函数等待event Action<T>来自另一个类和另一个线程的第一个实例,并在我的线程上处理它,允许等待被 timeout 或CancellationToken.

我想创建一个可以重用的通用函数。我设法创建了几个(我认为)我需要的选项,但两者似乎都比我想象的要复杂得多。

用法

为了清楚起见,此函数的示例使用如下所示,其中serialDevice在单独的线程上吐出事件:

var eventOccurred = Helper.WaitForSingleEvent<StatusPacket>(
    cancellationToken,
    statusPacket => OnStatusPacketReceived(statusPacket),
    a => serialDevice.StatusPacketReceived += a,
    a => serialDevice.StatusPacketReceived -= a,
    5000,
    () => serialDevice.RequestStatusPacket());

选项 1—ManualResetEventSlim

这个选项还不错,但是Dispose处理ManualResetEventSlim起来比看起来应该的要复杂。它使 ReSharper 适合我在闭包中访问修改/处置的东西,而且真的很难理解,所以我什至不确定它是否正确。也许我缺少一些可以清理它的东西,这将是我的偏好,但我没有立即看到它。这是代码。

public static bool WaitForSingleEvent<TEvent>(this CancellationToken token, Action<TEvent> handler, Action<Action<TEvent>> subscribe, Action<Action<TEvent>> unsubscribe, int msTimeout, Action initializer = null)
{
    var eventOccurred = false;
    var eventResult = default(TEvent);
    var o = new object();
    var slim = new ManualResetEventSlim();
    Action<TEvent> setResult = result => 
    {
        lock (o) // ensures we get the first event only
        {
            if (!eventOccurred)
            {
                eventResult = result;
                eventOccurred = true;
                // ReSharper disable AccessToModifiedClosure
                // ReSharper disable AccessToDisposedClosure
                if (slim != null)
                {
                    slim.Set();
                }
                // ReSharper restore AccessToDisposedClosure
                // ReSharper restore AccessToModifiedClosure
            }
        }
    };
    subscribe(setResult);
    try
    {
        if (initializer != null)
        {
            initializer();
        }
        slim.Wait(msTimeout, token);
    }
    finally // ensures unsubscription in case of exception
    {
        unsubscribe(setResult);
        lock(o) // ensure we don't access slim
        {
            slim.Dispose();
            slim = null;
        }
    }
    lock (o) // ensures our variables don't get changed in middle of things
    {
        if (eventOccurred)
        {
            handler(eventResult);
        }
        return eventOccurred;
    }
}

选项 2 — 不带WaitHandle

这里的WaitForSingleEvent功能更干净。我可以使用ConcurrentQueue,因此甚至不需要锁。但我只是不喜欢轮询功能Sleep,而且我看不出用这种方法有什么办法。我想传入 aWaitHandle而不是 aFunc<bool>来清理Sleep,但是第二次这样做我Dispose又要把整个烂摊子清理干净。

public static bool WaitForSingleEvent<TEvent>(this CancellationToken token, Action<TEvent> handler, Action<Action<TEvent>> subscribe, Action<Action<TEvent>> unsubscribe, int msTimeout, Action initializer = null)
{
    var q = new ConcurrentQueue<TEvent>();
    subscribe(q.Enqueue);
    try
    {
        if (initializer != null)
        {
            initializer();
        }
        token.Sleep(msTimeout, () => !q.IsEmpty);
    }
    finally // ensures unsubscription in case of exception
    {
        unsubscribe(q.Enqueue);
    }
    TEvent eventResult;
    var eventOccurred = q.TryDequeue(out eventResult);
    if (eventOccurred)
    {
        handler(eventResult);
    }
    return eventOccurred;
}

public static void Sleep(this CancellationToken token, int ms, Func<bool> exitCondition)
{
    var start = DateTime.Now;
    while ((DateTime.Now - start).TotalMilliseconds < ms && !exitCondition())
    {
        token.ThrowIfCancellationRequested();
        Thread.Sleep(1);
    }
}

问题

我并不特别关心这些解决方案中的任何一个,我也不能 100% 确定它们中的任何一个都是 100% 正确的。这些解决方案中的任何一个是否比另一个更好(惯用性、效率等),还是有更简单的方法或内置功能来满足我在这里需要做的事情?

更新:迄今为止的最佳答案

以下解决方案的修改TaskCompletionSource。无需长时间关闭、锁或任何需要的东西。看起来很简单。这里有什么错误吗?

public static bool WaitForSingleEvent<TEvent>(this CancellationToken token, Action<TEvent> onEvent, Action<Action<TEvent>> subscribe, Action<Action<TEvent>> unsubscribe, int msTimeout, Action initializer = null)
{
    var tcs = new TaskCompletionSource<TEvent>();
    Action<TEvent> handler = result => tcs.TrySetResult(result);
    var task = tcs.Task;
    subscribe(handler);
    try
    {
        if (initializer != null)
        {
            initializer();
        }
        task.Wait(msTimeout, token);
    }
    finally
    {
        unsubscribe(handler);
        // Do not dispose task http://blogs.msdn.com/b/pfxteam/archive/2012/03/25/10287435.aspx
    }
    if (task.Status == TaskStatus.RanToCompletion)
    {
        onEvent(task.Result);
        return true;
    }
    return false;
}

更新 2:另一个很好的解决方案

事实证明,它的BlockingCollection工作原理与此类似ConcurrentQueue,但也有接受超时和取消令牌的方法。这个解决方案的一个好处是它可以更新为WaitForNEvents相当容易:

public static bool WaitForSingleEvent<TEvent>(this CancellationToken token, Action<TEvent> handler, Action<Action<TEvent>> subscribe, Action<Action<TEvent>> unsubscribe, int msTimeout, Action initializer = null)
{
    var q = new BlockingCollection<TEvent>();
    Action<TEvent> add = item => q.TryAdd(item);
    subscribe(add);
    try
    {
        if (initializer != null)
        {
            initializer();
        }
        TEvent eventResult;
        if (q.TryTake(out eventResult, msTimeout, token))
        {
            handler(eventResult);
            return true;
        }   
        return false;
    }
    finally
    {
        unsubscribe(add);
        q.Dispose();
    }
}
4

4 回答 4

6

您可以使用TaskCompletetionSource创建一个Task您可以标记为已完成或已取消的。这是特定事件的可能实现:

public Task WaitFirstMyEvent(Foo target, CancellationToken cancellationToken)
{
    var tcs = new TaskCompletionSource<object>();
    Action handler = null;
    var registration = cancellationToken.Register(() =>
    {
        target.MyEvent -= handler;
        tcs.TrySetCanceled();
    });
    handler = () =>
    {
        target.MyEvent -= handler;
        registration.Dispose();
        tcs.TrySetResult(null);
    };
    target.MyEvent += handler;
    return tcs.Task;
}

在 C# 5 中,您可以像这样使用它:

private async Task MyMethod()
{
    ...
    await WaitFirstMyEvent(foo, cancellationToken);
    ...
}

如果要同步等待事件,也可以使用Wait方法:

private void MyMethod()
{
    ...
    WaitFirstMyEvent(foo, cancellationToken).Wait();
    ...
}

这是一个更通用的版本,但它仍然只适用于带有Action签名的事件:

public Task WaitFirstEvent(
    Action<Action> subscribe,
    Action<Action> unsubscribe,
    CancellationToken cancellationToken)
{
    var tcs = new TaskCompletionSource<object>();
    Action handler = null;
    var registration = cancellationToken.Register(() =>
    {
        unsubscribe(handler);
        tcs.TrySetCanceled();
    });
    handler = () =>
    {
        unsubscribe(handler);
        registration.Dispose();
        tcs.TrySetResult(null);
    };
    subscribe(handler);
    return tcs.Task;
}

你可以像这样使用它:

await WaitFirstEvent(
        handler => foo.MyEvent += handler,
        handler => foo.MyEvent -= handler,
        cancellationToken);

如果您希望它与其他事件签名(例如EventHandler)一起使用,则必须创建单独的重载。我认为没有一种简单的方法可以使它适用于任何签名,尤其是因为参数的数量并不总是相同的。

于 2013-07-14T01:02:20.023 回答
2

您可以使用 Rx 将事件转换为 observable,然后转换为任务,最后使用令牌/超时等待该任务。

与任何现有解决方案相比,它的一个优点是它调用unsubscribe事件的线程,确保您的处理程序不会被调用两次。tcs.TrySetResult(在您的第一个解决方案中,您可以通过代替 来解决此问题tcs.SetResult,但摆脱“TryDoSomething”并确保 DoSomething 始终有效总是很好的)。

另一个优点是代码的简单性。它本质上是一条线。所以你甚至不需要一个独立的函数。您可以内联它,以便更清楚您的代码究竟做了什么,并且您可以在不需要大量可选参数的情况下对主题进行更改(例如您的 optional initializer,或允许等待 N 个事件,或在以下情况下取消超时/取消)它们不是必需的)。当它完成时,你将同时拥有bool返回值实际result范围,如果这有用的话。

using System.Reactive.Linq;
using System.Reactive.Threading.Tasks;
...
public static bool WaitForSingleEvent<TEvent>(this CancellationToken token, Action<TEvent> onEvent, Action<Action<TEvent>> subscribe, Action<Action<TEvent>> unsubscribe, int msTimeout, Action initializer = null) {
    var task = Observable.FromEvent(subscribe, unsubscribe).FirstAsync().ToTask();
    if (initializer != null) {
        initializer();
    }
    try {
        var finished = task.Wait(msTimeout, token);
        if (finished) onEvent(task.Result);
        return finished;
    } catch (OperationCanceledException) { return false; }
}
于 2016-01-09T01:31:49.780 回答
0

非常感谢!为了帮助其他人理解......(可能显示带有点击操作处理程序代码的串行设备代码)

您还可以添加一个泛型类型约束,添加类似

 where TEvent : EventArgs

在我的情况下,我还需要“服务员”中的事件结果,
所以我改变了签名,比如
(在通用对象上又快又丑……)

 public static bool WaitForSingleEventWithResult<TEvent, TObjRes>(
            this CancellationToken token,
            Func<TEvent, TObjRes> onEvent,
             ...

以这种方式调用它

        var ct = new CancellationToken();
        object result;
        bool eventOccurred = ct.WaitForSingleEventWithResult<MyEventArgs, object>(
            onEvent: statusPacket => result = this.OnStatusPacketReceived(statusPacket),
            subscribe: sub => cp.StatusPacketReceived_Action += sub,
            unsubscribe: unsub => cp.StatusPacketReceived_Action -= unsub,
            msTimeout: 5 * 1000,
            initializer: /*() => serialDevice.RequestStatusPacket()*/null);

无论如何...非常感谢!

于 2019-05-07T10:11:36.980 回答
0

为什么不直接使用 ManualResetEventSlim.Wait (int millisecondsTimeout, CancellationToken cancellationToken)

于 2019-05-31T14:07:22.197 回答