Giga, EventHub

Giga, EventHub
F:\Projects\VS\26\EventHub\EventHubs\EventHubs.slnx
https://giga.chat/link/gcsYZSyuQh
Простой пример реализации события (Event) в C#, которое позволяет подписываться на уведомления о событии и получать оповещения при наступлении определенного условия:
csharp
Копировать
using System;

// Класс с событием
public class MyClassWithEvent
{
    // Объявляем делегат для нашего события
    public delegate void SimpleEventHandler(string message);
    
    // Создаем событие на основе делегата
    public event SimpleEventHandler OnSimpleEvent;

    // Метод, инициирующий событие
    public void RaiseEvent()
    {
        if (OnSimpleEvent != null)
            OnSimpleEvent("Событие произошло!");
    }
}

class Program
{
    static void Main()
    {
        var myInstance = new MyClassWithEvent();
        
        // Подписываемся на событие
        myInstance.OnSimpleEvent += HandleSimpleEvent;
        
        // Вызываем метод, который поднимает событие
        myInstance.RaiseEvent();
    }

    // Метод-обработчик события
    private static void HandleSimpleEvent(string message)
    {
        Console.WriteLine(message); // Выведет сообщение, переданное в обработчике
    }
}
Как работает этот код?
Делегат: Мы объявляем делегат SimpleEventHandler, который определяет сигнатуру метода-обработчика события.
csharp
Копировать
public delegate void SimpleEventHandler(string message);
Событие: Далее мы создаем само событие на основе делегата.
csharp
Копировать
public event SimpleEventHandler OnSimpleEvent;
Метод инициации события: Внутри класса создаётся метод RaiseEvent(), который проверяет наличие подписчиков и вызывает событие, передавая аргумент.
csharp
Копировать
public void RaiseEvent()
{
    if (OnSimpleEvent != null)
        OnSimpleEvent("Событие произошло!");
}
Подписка на событие: Из основной программы подпишемся на событие, добавив обработчик методом.
csharp
Копировать
myInstance.OnSimpleEvent += HandleSimpleEvent;
Обработка события: Когда событие поднимается, вызванный метод выводит сообщение в консоль.
Таким образом, простая реализация события в C# позволяет эффективно управлять асинхронностью действий и уведомлять заинтересованные компоненты системы о произошедших событиях.

Вопрос

Простой EventHubEventHub у нас простой Simple, EventHub// Interfaces.cspublic interface IPublisherService{    void Publish(string ticker, string message);}public interface ISubscriberService{    void Subscribe(string ticker, Action<string> handler);}
public class EventHub : Dictionary<string, List<Action<string>>>{    public void AddHandler(string ticker, Action<string> handler)    {        if (!ContainsKey(ticker))            this[ticker] = new List<Action<string>>();        this[ticker].Add(handler);    }
Копироватьpublic void InvokeHandlers(string ticker, string message){    if (TryGetValue(ticker, out var handlers))    {        foreach (var handler in handlers)            handler(message);    }}}

Ответ

Вот упрощенный вариант простого EventHub на C#, используя интерфейсы и классы, реализующие базовую функциональность публикации сообщений и подписки на события.
Простое определение интерфейсов
Определяются два основных интерфейса:
IPublisherService — интерфейс для издателя, позволяющий опубликовать сообщение.
ISubscriberService — интерфейс для подписчика, позволяющий подписаться на получение сообщений.
csharp
Копировать
// Интерфейсы
public interface IPublisherService
{
    void Publish(string ticker, string message);
}

public interface ISubscriberService
{
    void Subscribe(string ticker, Action<string> handler);
}
Реализация EventHub
Реализуем класс EventHub, который выступает посредником между издателями и подписчиками. Он хранит список слушателей для каждого тикера и обрабатывает события публикации и подписку.
csharp
Копировать
// Основной класс EventHub
public class EventHub : IPublisherService, ISubscriberService
{
    // Хранение тикеров и соответствующих обработчиков
    private readonly Dictionary<string, List<Action<string>>> _handlers =
        new Dictionary<string, List<Action<string>>>();

    // Добавление нового обработчика для конкретного тикера
    public void AddHandler(string ticker, Action<string> handler)
    {
        lock (_handlers)
        {
            if (!_handlers.ContainsKey(ticker))
                _handlers.Add(ticker, new List<Action<string>>());
            
            _handlers[ticker].Add(handler);
        }
    }

    // Выполнение всех зарегистрированных обработчиков для указанного тикера
    public void InvokeHandlers(string ticker, string message)
    {
        lock (_handlers)
        {
            if (_handlers.TryGetValue(ticker, out var handlers))
            {
                foreach (var handler in handlers.ToList())
                    handler.Invoke(message);
            }
        }
    }

    // Реализация методов интерфейсов
    public void Publish(string ticker, string message)
    {
        InvokeHandlers(ticker, message);
    }

    public void Subscribe(string ticker, Action<string> handler)
    {
        AddHandler(ticker, handler);
    }
}
Использование EventHub
Пример простейшего использования EventHub в приложении:
csharp
Копировать
class Program
{
    static void Main()
    {
        // Создание экземпляра EventHub
        var hub = new EventHub();

        // Регистрация слушателя для тикера "NewsTicker"
        hub.Subscribe("NewsTicker", msg => Console.WriteLine($"Received news: {msg}"));

        // Публикация сообщения
        hub.Publish("NewsTicker", "Hello from the publisher");
    }
}
Объяснение:
Интерфейсы: Определяют методы для публикаций и подписки.
EventHub: Хранилище отображающих отношения между именами ("тикерами") и списком обработчиков событий.
Методы: 
Subscribe: добавляет новый обработчик для конкретного тикера.
Publish: отправляет сообщение всем зарегистрированным обработчикам для заданного тикера.
Lock'и: Используются для синхронизации потоков, предотвращают одновременное изменение состояния списка обработчиков несколькими потоками.
Это простое решение хорошо подходит для небольших приложений и прототипирования, позволяя легко расширять систему публикаций-подписок.

Вопрос

Как реализовать простую публикацию и подписку на разные тикеры?Какие преимущества имеет использование EventHub в C#?Как избежать гонки потоков при использовании EventHub?

Ответ

Реализация простой публикации и подписки на разные тикеры в EventHub
Шаг 1: Определение структуры хранения обработчиков
Используется словарь (Dictionary), где ключом является название тикера, а значением — список обработчиков событий типа Action<string>:
csharp
Копировать
private readonly Dictionary<string, List<Action<string>>> _handlers = new Dictionary<string, List<Action<string>>>();
Шаг 2: Реализация механизма подписки и публикации
Класс EventHub реализует механизм добавления новых обработчиков и отправки сообщений всем подписанным обработчикам:
csharp
Копировать
public class EventHub
{
    private readonly Dictionary<string, List<Action<string>>> _handlers = new Dictionary<string, List<Action<string>>>();

    // Добавляет обработчик к определенному тикеру
    public void Subscribe(string ticker, Action<string> handler)
    {
        lock (_handlers)
        {
            if (!_handlers.ContainsKey(ticker))
                _handlers[ticker] = new List<Action<string>>();
            
            _handlers[ticker].Add(handler);
        }
    }

    // Отправляет сообщение всем обработчикам, связанным с данным тикером
    public void Publish(string ticker, string message)
    {
        lock (_handlers)
        {
            if (_handlers.TryGetValue(ticker, out var handlers))
            {
                foreach (var handler in handlers.ToList()) // Чтобы предотвратить модификацию коллекции во время итерации
                    handler.Invoke(message);
            }
        }
    }
}
Преимущества использования EventHub в C#
Использование EventHub обладает рядом преимуществ:
Разделение ответственности
Publisher и Subscriber отделены друг от друга. Издатель публикует сообщения, не зная ничего о подписчиках, а подписчики получают обновления независимо от конкретных деталей издательского процесса.
Масштабируемость
Система легко масштабируется путем добавления новых тикеров и обработчиков. Это удобно, если приложение развивается и требует больше каналов связи.
Модульность и простота тестирования
Каждый компонент (Publisher, Subscriber, EventHub) легко тестируется отдельно, что улучшает качество разработки и снижает риск ошибок.
Легкость расширения
Можно добавлять новые типы сообщений и тикеров без изменения существующего кода, соблюдая принцип открытости-закрытости SOLID.
Безопасность и изоляция компонентов
Уменьшается зависимость между различными частями приложения, что повышает устойчивость всей архитектуры.
Предотвращение гонок потоков в EventHub
Чтобы избежать проблем с конкурентным доступом к общим ресурсам (например, списку обработчиков), рекомендуется применять механизмы блокировки потока. Вот несколько подходов:
Синхронизация с использованием оператора lock
Самый простой способ — обернуть критический участок кода в оператор lock. Этот метод гарантирует, что одновременно только один поток сможет изменять состояние коллекций обработчиков.
csharp
Копировать
lock (_handlers)
{
    // Изменяем коллекцию внутри замка
}
Создание отдельной копии списка обработчиков
Если количество обработчиков велико, и их добавление занимает много времени, лучше создать копию списка перед выполнением итерации, чтобы исключить исключение Collection was modified.
csharp
Копировать
foreach (var handler in handlers.ToList())
{
    handler.Invoke(message);
}
Использование иммутабельных структур данных
Вместо обычных списков можно использовать иммутабельные коллекции (immutable collections). Они обеспечивают защиту от изменений и помогают избегать ситуаций, когда коллекция изменяется параллельно во время обработки.
Thread-safe версии классов
Некоторые встроенные коллекции .NET имеют thread-safe версии, например, ConcurrentDictionary<TK,TV>. Однако стоит учитывать, что такие коллекции часто менее эффективны, особенно при большом количестве операций записи.
Итоговая рекомендация
Для большинства случаев достаточно простых механизмов защиты вроде lock. Для высоконагруженных систем стоит рассмотреть использование специализированных инструментов для управления параллелизмом и синхронизацией.

Вопрос

Как добавить поддержку нескольких тикеров в EventHub?Какие альтернативы использованию блокировки lock существуют для предотвращения гонок потоков в EventHub?Почему важно разделять Publisher и Subscriber в архитектуре EventHub?

Ответ

Поддержка нескольких тикеров в EventHub
Поддержка нескольких тикеров реализуется простым добавлением ключа-тикера в словарь, который связывает каждый тикер с набором обработчиков событий. Рассмотрим подробнее шаги для поддержки множественных тикеров:
Хранение состояний по каждому тикеру
В качестве хранилища используется словарь (Dictionary<string, List<Action<string>>>), где ключом служит тикер, а значением — список обработчиков.
csharp
Копировать
private readonly Dictionary<string, List<Action<string>>> _handlers = new Dictionary<string, List<Action<string>>>();
Механизм подписки
Каждого подписчика привязывают к определённому тикеру, сохраняя обработчик в списке для соответствующего тикера.
csharp
Копировать
public void Subscribe(string ticker, Action<string> handler)
{
    lock (_handlers)
    {
        if (!_handlers.ContainsKey(ticker))
            _handlers[ticker] = new List<Action<string>>();
        
        _handlers[ticker].Add(handler);
    }
}
Отправка сообщений
Сообщения отправляются именно тем обработчикам, которые связаны с указанным тикером.
csharp
Копировать
public void Publish(string ticker, string message)
{
    lock (_handlers)
    {
        if (_handlers.TryGetValue(ticker, out var handlers))
        {
            foreach (var handler in handlers.ToList())
                handler.Invoke(message);
        }
    }
}
Теперь каждый тикер ведет собственный список обработчиков, и каждое опубликованное сообщение доставляется только нужным подписчикам.
Альтернативы механизму блокировки (lock)
Альтернативы блокировке с использованием lock включают:
Immutability (неизменяемые коллекции):
Вы можете использовать неизменяемые коллекции, например, ImmutableDictionary или ImmutableList, чтобы обеспечить безопасность многопоточности. Такие коллекции позволяют автоматически создавать защищённые от модификации представления данных.
csharp
Копировать
using System.Collections.Immutable;

private ImmutableDictionary<string, ImmutableList<Action<string>>> _handlers = ImmutableDictionary.Create<string, ImmutableList<Action<string>>>();

public void Subscribe(string ticker, Action<string> handler)
{
    _handlers = _handlers.SetItem(
        ticker,
        _handlers.GetValueOrDefault(ticker)?.Add(handler) ?? ImmutableList.Create(handler));
}
Concurrent Collections (thread-safe коллекции):
Используйте специализированные коллекции, предназначенные для безопасной работы в многопоточной среде, такие как ConcurrentDictionary<TKey, TValue>. Эти коллекции внутренне используют оптимизированные алгоритмы блокировки, обеспечивая высокую производительность даже при частых операциях чтения-записи.
csharp
Копировать
private ConcurrentDictionary<string, List<Action<string>>> _handlers = new ConcurrentDictionary<string, List<Action<string>>>();

public void Subscribe(string ticker, Action<string> handler)
{
    _handlers.AddOrUpdate(ticker, key => new List<Action<string>> { handler }, (key, oldVal) =>
    {
        oldVal.Add(handler);
        return oldVal;
    });
}
ReaderWriterLockSlim:
Эта структура обеспечивает раздельные блокировки для читателей и писателей, минимизируя задержки и улучшая производительность при работе с большими объемами данных.
csharp
Копировать
private readonly ReaderWriterLockSlim _rwLock = new ReaderWriterLockSlim();

public void Subscribe(string ticker, Action<string> handler)
{
    _rwLock.EnterWriteLock();
    try
    {
        if (!_handlers.ContainsKey(ticker))
            _handlers[ticker] = new List<Action<string>>();
       
        _handlers[ticker].Add(handler);
    }
    finally
    {
        _rwLock.ExitWriteLock();
    }
}
Эти подходы предоставляют гибкость выбора стратегии синхронизации, исходя из требований производительности и нагрузки.
Почему важно разделять Publisher и Subscriber в архитектуре EventHub?
Разделение обязанностей между Publisher и Subscriber приносит ряд важных преимуществ:
Независимость компонентов:
Разделяя ответственность, вы уменьшаете связь между разными частями системы. Например, Publisher не зависит от конкретных подписчиков, а подписчики не знают, откуда приходят сообщения. Это упрощает разработку и тестирование отдельных частей системы.
Масштабируемость:
Возможность разделения компонентов облегчает горизонтальное масштабирование. Так, отдельные сервисы могут обрабатывать публикации и подписки независимо друг от друга, увеличивая общую пропускную способность системы.
Повышение устойчивости:
Независимый подход уменьшает вероятность сбоев в одной части системы влиять на остальные компоненты. Например, ошибка в одном подписчике не должна приводить к отказу всего сервиса публикации.
Улучшенная модульность:
Благодаря четкому разделению ролей проще проводить рефакторинг и вносить улучшения в отдельные модули без риска нарушить работу остальных компонентов.
Упрощённая обработка сложных сценариев:
Архитектура EventHub идеально подходит для асинхронных взаимодействий и больших распределённых систем, где важна скорость реакции на события и надежность доставки сообщений.
В итоге разделение обязанностей способствует созданию устойчивой, производительной и удобной для сопровождения архитектуры.

Вопрос

using System.Collections.Immutable;
private ImmutableDictionary<string, ImmutableList<Action<string>>> _handlers = ImmutableDictionary.Create<string, ImmutableList<Action<string>>>();
public void Subscribe(string ticker, Action<string> handler){    _handlers = _handlers.SetItem(        ticker,        _handlers.GetValueOrDefault(ticker)?.Add(handler) ?? ImmutableList.Create(handler));}
private ConcurrentDictionary<string, List<Action<string>>> _handlers = new ConcurrentDictionary<string, List<Action<string>>>();
public void Subscribe(string ticker, Action<string> handler){    _handlers.AddOrUpdate(ticker, key => new List<Action<string>> { handler }, (key, oldVal) =>    {        oldVal.Add(handler);        return oldVal;    });}
private readonly ReaderWriterLockSlim _rwLock = new ReaderWriterLockSlim();
public void Subscribe(string ticker, Action<string> handler){    _rwLock.EnterWriteLock();    try    {        if (!_handlers.ContainsKey(ticker))            _handlers[ticker] = new List<Action<string>>();
Копировать
    _handlers[ticker].Add(handler);
}
finally
{
    _rwLock.ExitWriteLock();
}
}
Напиши три полных реализаций для всех трех различных вариантов EventHub

Ответ

Ниже приведены полные реализации трёх различных вариантов EventHub с поддержкой нескольких тикеров и безопасностью многопоточности.
Вариант №1: Иммутабельная коллекция (ImmutableDictionary)
Этот подход основан на неизменяемых коллекциях, обеспечивающих автоматическое создание копий при каждом изменении.
csharp
Копировать
using System;
using System.Collections.Generic;
using System.Collections.Immutable;

public class EventHub
{
    private ImmutableDictionary<string, ImmutableList<Action<string>>> _handlers = ImmutableDictionary.Create<string, ImmutableList<Action<string>>>();

    // Подписывается на указанный тикер
    public void Subscribe(string ticker, Action<string> handler)
    {
        _handlers = _handlers.SetItem(
            ticker,
            _handlers.GetValueOrDefault(ticker)?.Add(handler) ?? ImmutableList.Create(handler));
    }

    // Отписывается от тикера
    public bool Unsubscribe(string ticker, Action<string> handler)
    {
        if (_handlers.TryGetValue(ticker, out var list))
        {
            var updatedList = list.Remove(handler);
            if (updatedList.Count > 0 || !list.Any())
            {
                _handlers = _handlers.SetItem(ticker, updatedList);
                return true;
            }
        }
        return false;
    }

    // Публикует сообщение для указанного тикера
    public void Publish(string ticker, string message)
    {
        if (_handlers.TryGetValue(ticker, out var handlers))
        {
            foreach (var handler in handlers)
                handler.Invoke(message);
        }
    }
}

class Program
{
    static void Main()
    {
        var hub = new EventHub();
        hub.Subscribe("Stocks", msg => Console.WriteLine($"Stock update: {msg}"));
        hub.Subscribe("Weather", msg => Console.WriteLine($"Weather report: {msg}"));

        hub.Publish("Stocks", "Apple stock is rising.");
        hub.Publish("Weather", "It will rain tomorrow.");
    }
}
Вариант №2: Thread-Safe Коллекция (ConcurrentDictionary)
Здесь используется специализированная коллекция, предназначенная для работы в многопоточных средах, что устраняет необходимость вручную ставить блокировки.
csharp
Копировать
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;

public class EventHub
{
    private ConcurrentDictionary<string, List<Action<string>>> _handlers = new ConcurrentDictionary<string, List<Action<string>>>();

    // Подписывается на указанный тикер
    public void Subscribe(string ticker, Action<string> handler)
    {
        _handlers.AddOrUpdate(ticker, key => new List<Action<string>> { handler },
                              (key, oldVal) =>
                              {
                                  oldVal.Add(handler);
                                  return oldVal;
                              });
    }

    // Отписывается от тикера
    public bool Unsubscribe(string ticker, Action<string> handler)
    {
        if (_handlers.TryGetValue(ticker, out var list))
        {
            list.Remove(handler);
            return true;
        }
        return false;
    }

    // Публикует сообщение для указанного тикера
    public void Publish(string ticker, string message)
    {
        if (_handlers.TryGetValue(ticker, out var handlers))
        {
            foreach (var handler in handlers.ToList())
                handler.Invoke(message);
        }
    }
}

class Program
{
    static void Main()
    {
        var hub = new EventHub();
        hub.Subscribe("Stocks", msg => Console.WriteLine($"Stock update: {msg}"));
        hub.Subscribe("Weather", msg => Console.WriteLine($"Weather report: {msg}"));

        hub.Publish("Stocks", "Google stock dropped.");
        hub.Publish("Weather", "Sunny day ahead.");
    }
}
Вариант №3: Чтение-запись с использованием ReaderWriterLockSlim
Эта техника применяется, когда требуется разделить операции чтения и записи для повышения производительности.
csharp
Копировать
using System;
using System.Collections.Generic;
using System.Threading;

public class EventHub
{
    private readonly ReaderWriterLockSlim _rwLock = new ReaderWriterLockSlim();
    private readonly Dictionary<string, List<Action<string>>> _handlers = new Dictionary<string, List<Action<string>>>();

    // Подписывается на указанный тикер
    public void Subscribe(string ticker, Action<string> handler)
    {
        _rwLock.EnterWriteLock();
        try
        {
            if (!_handlers.ContainsKey(ticker))
                _handlers[ticker] = new List<Action<string>>();
            
            _handlers[ticker].Add(handler);
        }
        finally
        {
            _rwLock.ExitWriteLock();
        }
    }

    // Отписывается от тикера
    public bool Unsubscribe(string ticker, Action<string> handler)
    {
        _rwLock.EnterWriteLock();
        try
        {
            if (_handlers.TryGetValue(ticker, out var list))
            {
                list.Remove(handler);
                return true;
            }
        }
        finally
        {
            _rwLock.ExitWriteLock();
        }
        return false;
    }

    // Публикует сообщение для указанного тикера
    public void Publish(string ticker, string message)
    {
        _rwLock.EnterReadLock();
        try
        {
            if (_handlers.TryGetValue(ticker, out var handlers))
            {
                foreach (var handler in handlers.ToList())
                    handler.Invoke(message);
            }
        }
        finally
        {
            _rwLock.ExitReadLock();
        }
    }
}

class Program
{
    static void Main()
    {
        var hub = new EventHub();
        hub.Subscribe("Stocks", msg => Console.WriteLine($"Stock update: {msg}"));
        hub.Subscribe("Weather", msg => Console.WriteLine($"Weather report: {msg}"));

        hub.Publish("Stocks", "Microsoft's stocks are stable.");
        hub.Publish("Weather", "Light breeze expected today.");
    }
}
Заключение
Каждый из представленных способов решает проблему безопасности многопоточности в EventHub. Выбор конкретной реализации зависит от контекста вашего проекта:
Иммутабельность: проста в реализации, удобна для относительно небольшого числа обновлений.
Concurrent-коллекции: подходят для интенсивных операций ввода-вывода и большого количества записей.
ReaderWriterLockSlim: оптимальный выбор, если преобладают операции чтения над операциями записи.
Выбирайте тот подход, который наилучшим образом соответствует вашим требованиям к производительности и нагрузкам вашей системы.