C# 在单独的线程上引发事件

Raising events on separate thread

提问by Hitesh P

I am developing a component which needs to process the live feed and broadcast the data to the listeners in pretty fast manner ( with about 100 nano second level accuracy, even less than that if I can do that) Currently I am raising an event from my code which subscriber can subscribe to. However because in C# event handlers run on the same thread which raises the event, my thread which raises the event will be blocked until all subscribers finish processing the event. I do not have control on subscribers' code, so they can possibly do any time consuming operations in event handler, which may block the thread which is broadcasting.

我正在开发一个组件，它需要处理实时馈送并将数据以非常快的方式广播给听众（大约 100 纳秒级的精度，如果我能做到的话，甚至比这个精度还要低）目前我正在从我的订阅者可以订阅的代码。但是，因为在 C# 事件处理程序中运行在引发事件的同一个线程上，我引发事件的线程将被阻塞，直到所有订阅者完成处理事件。我无法控制订阅者的代码，因此他们可能会在事件处理程序中执行任何耗时的操作，这可能会阻塞正在广播的线程。

What can I do so that I can broadcast the data to other subscribers but can still broadcast the stuff quite fast??

我该怎么做才能将数据广播给其他订阅者，但仍然可以很快地广播这些内容？？

回答by Iddillian

It seems like you are looking for tasks. The following is an extension method i wrote for my job that can asynchronously invokes an event so that every event handler is on their own thread. I can't comment on its speed since that has never been a requirement for me.

看起来您正在寻找任务。以下是我为我的工作编写的扩展方法，它可以异步调用一个事件，以便每个事件处理程序都在自己的线程上。我无法评论它的速度，因为这从来不是我的要求。

---

---

UPDATE

更新

Based on the comments i adjusted it so that only one task is created to call all of the subscribers

根据评论，我对其进行了调整，以便只创建一个任务来调用所有订阅者

/// <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);
            }
        });
    }
}

回答by dss539

100 ns is a very tough target to hit. I believe it will take a deep understanding of what you're doing and why to hit that kind of performance.

100 ns 是一个很难达到的目标。我相信这需要深入了解你在做什么以及为什么要达到这种表现。

However, asynchronously invoking event subscribers is pretty easy to solve. It's already answered hereby, who else, Jon Skeet.

然而，异步调用事件订阅者很容易解决。已经在这里回答了，还有谁，Jon Skeet。

foreach (MyDelegate action in multicast.GetInvocationList())
{
    action.BeginInvoke(...);
}

edit:I should also mention that you need to be running on a real-time operating systemto give tight performance guarantees to your users.

编辑：我还应该提到您需要在实时操作系统上运行才能为您的用户提供严格的性能保证。

回答by deepee1

I can't speak to if this will reliably meet the 100ns requirement but here's an alternative where you'd provide the end user with a way to provide you a ConcurrentQueue that you would fill and they could listen to on a separate thread.

我无法确定这是否能可靠地满足 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; }
}

Since given the fact that you can't hold up processing on other listeners this means multiple threads. Having dedicated listening threads on the listeners seems like a reasonable approach to try, and the concurrent queue seems like a decent delivery mechanism. In this implementation it's just constantly polling, but you could probably use thread signaling to reduce the cpu load using something like AutoResetEvent.

鉴于您无法阻止其他侦听器的处理，这意味着多线程。在侦听器上拥有专用的侦听线程似乎是一种合理的尝试方法，并发队列似乎是一种不错的交付机制。在此实现中，它只是不断轮询，但您可能可以使用线程信号来减少 cpu 负载，使用类似AutoResetEvent.

回答by Erwin Mayer

You can use these simple extension methods on your event handlers:

您可以在事件处理程序上使用这些简单的扩展方法：

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());
}

Usage:

用法：

public static Test() {
     myEventHandler.RaiseOnDifferentThread(null, EventArgs.Empty);
}

The cancellationTokenis necessary to guarantee StartNew()actually uses a different thread, as explained here.

该cancellationToken保证是必要的StartNew()实际使用不同的线程，如解释在这里。

回答by user191880

Signals and shared memory are very fast. You could send a separate signal to tell applications to read a message from a shared-memory location. Of course, the signal is still an event that your application has to consume on a high-priority thread if you want low latency. I would include a time-tag in the data so the receiver can compensate for the inevitable latency.

信号和共享内存非常快。您可以发送一个单独的信号来告诉应用程序从共享内存位置读取消息。当然，如果您想要低延迟，信号仍然是您的应用程序必须在高优先级线程上使用的事件。我会在数据中包含一个时间标签，以便接收器可以补偿不可避免的延迟。