如何在C#中等待单个事件，并设置超时和取消
因此，我的要求是让我的函数等待来自另一个类和另一个线程的event Action<T>的第一个实例，并在我的线程上处理它，允许等待被超时或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处理比它看起来应该的要混乱。我在闭包中访问修改过的/处理过的东西，这让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，而且我看不出这种方法有什么办法可以绕过它。我想传入一个WaitHandle而不是Func<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();
}
}回答 3
Stack Overflow用户
发布于 2013-07-14 09:02:20
您可以使用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)一起工作，则必须创建单独的重载。我认为没有一种简单的方法可以让它对任何签名起作用，特别是因为参数的数量并不总是相同的。
Stack Overflow用户
发布于 2019-05-07 18:11:37
非常感谢！为了帮助别人理解。(可能会显示带有hits操作处理程序代码的序列设备代码)
您还可以将泛型类型约束添加到如下内容中
where TEvent : EventArgs在我的例子中，我还需要"waiter“中事件外的结果。
所以我把签名改成了
(通用对象上的快速和丑陋...)
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);不管怎样..。非常感谢！
Stack Overflow用户
发布于 2019-05-31 22:07:22
为什么不直接使用ManualResetEventSlim.Wait (int millisecondsTimeout, CancellationToken cancellationToken)呢？
https://stackoverflow.com/questions/17635440
复制相似问题