我正在开发一个组件，它需要处理实时提要并以非常快的方式将数据广播给听众（精度约为 100 纳秒级，如果我能做到这一点，甚至会更低）目前我正在从我的订阅者可以订阅的代码。然而，因为在 C# 事件处理程序中运行在引发事件的同一线程上，所以引发事件的我的线程将被阻塞，直到所有订阅者完成处理事件。我无法控制订阅者的代码，因此他们可能会在事件处理程序中执行任何耗时的操作，这可能会阻塞正在广播的线程。
我该怎么做才能将数据广播给其他订阅者，但仍然可以很快地广播这些内容？
问问题
25583 次
5 回答
5
看起来你正在寻找任务。以下是我为我的工作编写的扩展方法，它可以异步调用事件，以便每个事件处理程序都在自己的线程上。我无法评论它的速度，因为这对我来说从来都不是要求。
更新
根据评论，我对其进行了调整，以便只创建一个任务来调用所有订阅者
/// <summary>
/// Extension method to safely encapsulate asynchronous event calls with checks
/// </summary>
/// <param name="evnt">The event to call</param>
/// <param name="sender">The sender of the event</param>
/// <param name="args">The arguments for the event</param>
/// <param name="object">The state information that is passed to the callback method</param>
/// <remarks>
/// This method safely calls the each event handler attached to the event. This method uses <see cref="System.Threading.Tasks"/> to
/// asynchronously call invoke without any exception handling. As such, if any of the event handlers throw exceptions the application will
/// most likely crash when the task is collected. This is an explicit decision since it is really in the hands of the event handler
/// creators to make sure they handle issues that occur do to their code. There isn't really a way for the event raiser to know
/// what is going on.
/// </remarks>
[System.Diagnostics.DebuggerStepThrough]
public static void AsyncSafeInvoke( this EventHandler evnt, object sender, EventArgs args )
{
// Used to make a temporary copy of the event to avoid possibility of
// a race condition if the last subscriber unsubscribes
// immediately after the null check and before the event is raised.
EventHandler handler = evnt;
if (handler != null)
{
// Manually calling all event handlers so that we could capture and aggregate all the
// exceptions that are thrown by any of the event handlers attached to this event.
var invocationList = handler.GetInvocationList();
Task.Factory.StartNew(() =>
{
foreach (EventHandler h in invocationList)
{
// Explicitly not catching any exceptions. While there are several possibilities for handling these
// exceptions, such as a callback, the correct place to handle the exception is in the event handler.
h.Invoke(sender, args);
}
});
}
}
于 2013-09-18T20:40:24.233 回答
4
您可以在事件处理程序上使用这些简单的扩展方法：
public static void Raise<T>(this EventHandler<T> handler, object sender, T e) where T : EventArgs {
if (handler != null) handler(sender, e);
}
public static void Raise(this EventHandler handler, object sender, EventArgs e) {
if (handler != null) handler(sender, e);
}
public static void RaiseOnDifferentThread<T>(this EventHandler<T> handler, object sender, T e) where T : EventArgs {
if (handler != null) Task.Factory.StartNewOnDifferentThread(() => handler.Raise(sender, e));
}
public static void RaiseOnDifferentThread(this EventHandler handler, object sender, EventArgs e) {
if (handler != null) Task.Factory.StartNewOnDifferentThread(() => handler.Raise(sender, e));
}
public static Task StartNewOnDifferentThread(this TaskFactory taskFactory, Action action) {
return taskFactory.StartNew(action: action, cancellationToken: new CancellationToken());
}
用法：
public static Test() {
myEventHandler.RaiseOnDifferentThread(null, EventArgs.Empty);
}
有cancellationToken必要保证StartNew()实际使用不同的线程，如here所述。
于 2013-11-28T11:47:22.123 回答
2
我不能说这是否能可靠地满足 100ns 的要求，但这里有一个替代方案，您可以为最终用户提供一种方法，为您提供一个您将填充的 ConcurrentQueue，他们可以在单独的线程上收听。
class Program
{
static void Main(string[] args)
{
var multicaster = new QueueMulticaster<int>();
var listener1 = new Listener(); //Make a couple of listening Q objects.
listener1.Listen();
multicaster.Subscribe(listener1);
var listener2 = new Listener();
listener2.Listen();
multicaster.Subscribe(listener2);
multicaster.Broadcast(6); //Send a 6 to both concurrent Queues.
Console.ReadLine();
}
}
//The listeners would run on their own thread and poll the Q like crazy.
class Listener : IListenToStuff<int>
{
public ConcurrentQueue<int> StuffQueue { get; set; }
public void Listen()
{
StuffQueue = new ConcurrentQueue<int>();
var t = new Thread(ListenAggressively);
t.Start();
}
void ListenAggressively()
{
while (true)
{
int val;
if(StuffQueue.TryDequeue(out val))
Console.WriteLine(val);
}
}
}
//Simple class that allows you to subscribe a Queue to a broadcast event.
public class QueueMulticaster<T>
{
readonly List<IListenToStuff<T>> _subscribers = new List<IListenToStuff<T>>();
public void Subscribe(IListenToStuff<T> subscriber)
{
_subscribers.Add(subscriber);
}
public void Broadcast(T value)
{
foreach (var listenToStuff in _subscribers)
{
listenToStuff.StuffQueue.Enqueue(value);
}
}
}
public interface IListenToStuff<T>
{
ConcurrentQueue<T> StuffQueue { get; set; }
}
鉴于您无法阻止其他侦听器的处理，这意味着多个线程。在侦听器上设置专用侦听线程似乎是一种合理的尝试方法，并发队列似乎是一种不错的交付机制。在这个实现中，它只是不断地轮询，但您可以使用线程信号来减少 cpu 负载，例如AutoResetEvent.
于 2013-09-18T22:30:45.130 回答
0
信号和共享内存非常快。您可以发送一个单独的信号来告诉应用程序从共享内存位置读取消息。当然，如果您想要低延迟，信号仍然是您的应用程序必须在高优先级线程上使用的事件。我会在数据中包含一个时间标签，以便接收器可以补偿不可避免的延迟。
于 2019-08-02T22:12:31.217 回答