Поднятие событий в отдельной ветке
Я разрабатываю компонент, который должен обрабатывать живой канал и транслировать данные слушателям довольно быстро (с точностью до 100 наномного уровня, даже меньше, чем если бы я мог это сделать). В настоящее время я поднимаю событие из моего кода, на который подписчик может подписаться. Однако, поскольку в обработчиках событий С# выполняется тот же поток, который вызывает событие, мой поток, который вызывает событие, будет заблокирован, пока все подписчики не закончат обработку события. У меня нет контроля над кодом подписчиков, поэтому они могут выполнять любые длительные операции в обработчике событий, которые могут блокировать поток, который транслируется.
Что я могу сделать, чтобы я мог транслировать данные другим подписчикам, но все же может транслировать материал довольно быстро??
Ответы
Ответ 1
100 нс - очень сложная цель попасть. Я полагаю, что это займет глубокое понимание того, что вы делаете, и почему ударить такую работу.
Однако асинхронное обращение к подписчикам на события довольно легко решить.
Он уже ответил здесь, кто еще, Джон Скит.
foreach (MyDelegate action in multicast.GetInvocationList())
{
action.BeginInvoke(...);
}
изменить
Я должен также упомянуть, что вам нужно работать в оперативной системе в режиме реального времени, чтобы предоставить пользователям гарантии производительности.
Ответ 2
Кажется, что вы ищете задачи. Ниже приведен метод расширения, который я написал для своей работы, который может асинхронно запускать событие, чтобы каждый обработчик событий находился в своем потоке. Я не могу прокомментировать его скорость, поскольку это никогда не требовало меня.
ОБНОВЛЕНИЕ
На основе комментариев я скорректировал его так, чтобы создать только одну задачу для вызова всех подписчиков
/// <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);
}
});
}
}
Ответ 3
Вы можете использовать эти простые методы расширения для своих обработчиков событий:
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()
фактически использует другой поток, как описано здесь.
Ответ 4
Я не могу говорить, если это будет надежно соответствовать требованию 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; }
}
Поскольку с учетом того факта, что вы не можете поддерживать обработку на других слушателях, это означает несколько потоков. Наличие выделенных потоков прослушивания у слушателей кажется разумным подходом к попытке, а параллельная очередь кажется приличным механизмом доставки. В этой реализации это просто постоянный опрос, но вы, вероятно, можете использовать сигнализацию потоков, чтобы уменьшить загрузку процессора, используя что-то вроде AutoResetEvent
.