ConcurrentBag и ConcurrentDictionary

1. Введение

ConcurrentBag<T> — это потокобезопасная, неупорядоченная коллекция. Её главная особенность и преимущество кроются в слове «Bag» (мешок), что подразумевает, что порядок элементов не гарантируется при извлечении. Это значит, что элемент, который вы извлечёте, может быть не тем, который вы ожидали на основе порядка добавления. Зато ConcurrentBag имеет уникальную оптимизацию, делающую её чрезвычайно быстрой в определённых сценариях.

Особенности ConcurrentBag

Отсутствие порядка: В отличие от очередей (FIFO) и стеков (LIFO), ConcurrentBag не обещает, что TryTake() вернёт вам элемент в каком-либо конкретном порядке относительно того, как он был добавлен. Это ключевое отличие.

Оптимизация для локального доступа (Thread-Local Storage): Основная причина существования ConcurrentBag — её производительность в сценариях, где поток, который добавил элемент, с высокой вероятностью будет тем же потоком, который его извлечёт.

Пример: ConcurrentBag — добавление и извлечение

using System.Collections.Concurrent;

ConcurrentBag<string> itemBag = new ConcurrentBag<string>();

// Добавление элементов
itemBag.Add("Пункт А");
itemBag.Add("Пункт Б");
itemBag.Add("Пункт В");

Console.WriteLine($"Элементов в мешке: {itemBag.Count}"); // Вывод: Элементов в мешке: 3

// Извлечение элементов (порядок не гарантируется!)
if (itemBag.TryTake(out string item1))
{
    Console.WriteLine($"Извлечено: {item1}"); // Может быть "Пункт В", "Пункт Б" или "Пункт А"
}

if (itemBag.TryTake(out string item2))
{
    Console.WriteLine($"Извлечено: {item2}");
}

Console.WriteLine($"Элементов осталось: {itemBag.Count}"); // Вывод: Элементов осталось: 1

Вы можете запустить этот код несколько раз и заметить, что порядок извлечённых элементов может меняться.

Методы Add(), TryTake()

Add(T item): используется для добавления элемента в ConcurrentBag. Операция потокобезопасна.

TryTake(out T item): попытка извлечения элемента из ConcurrentBag. Возвращает true, если элемент успешно извлечён, и false, если мешок пуст. Важно, что TryTake не блокирует поток.

2. Сценарии использования

ConcurrentBag — это не замена ConcurrentQueue или ConcurrentStack. Она сияет в специфических случаях:

Пулы объектов/ресурсов: когда у вас есть пул переиспользуемых объектов, и желательно, чтобы поток, который вернул объект, чаще всего брал его снова. Это снижает конкуренцию за общий ресурс.

Динамическое распределение задач в TPL: внутренняя работа таких конструкций как Parallel.ForEach и Parallel.For использует локальные мешки и механизм «work-stealing» для эффективного распределения работы.

Пул задач с ConcurrentBag и оптимизацией локальности

using System.Collections.Concurrent;
using System.Threading.Tasks;
using System.Threading;

ConcurrentBag<string> taskPool = new ConcurrentBag<string>();

// Заполняем пул начальными задачами
for (int i = 0; i < 10; i++)
{
    taskPool.Add($"Задача {i}");
}

void Worker()
{
    // Каждый поток пытается взять задачу
    while (taskPool.TryTake(out string task))
    {
        Console.WriteLine($"Поток {Thread.CurrentThread.ManagedThreadId}: Обрабатывает {task}");
        Thread.Sleep(50); // Имитация работы
    }
    Console.WriteLine($"Поток {Thread.CurrentThread.ManagedThreadId}: Завершил работу.");
}

// Запускаем несколько потоков-работников
// Task.Run(Worker);
// Task.Run(Worker);
// Task.Run(Worker);
// Thread.Sleep(1000); // Даем время на выполнение

В этом примере ConcurrentBag позволяет потокам эффективно брать задачи, минимизируя блокировки благодаря внутренней структуре.

Внутренняя механика

ConcurrentBag достигает высокой производительности за счёт использования потоко-локального хранилища (TLS). Когда поток добавляет элемент, он помещается в локальную для потока структуру. При TryTake() сначала читается локальная структура; если она пуста — выполняется «work-stealing» из других потоков или глобального пула. Это уменьшает конкуренцию и делает коллекцию отличным выбором, когда важна локальность доступа и не важен порядок.

3. Потокобезопасный словарь

ConcurrentDictionary<TKey, TValue> — одна из самых часто используемых потокобезопасных коллекций: высокопроизводительный словарь для безопасных операций добавления, чтения, обновления и удаления из нескольких потоков.

Обычный Dictionary<TKey, TValue> абсолютно не потокобезопасен. Любая запись (добавление/изменение/удаление) или даже чтение во время записи может привести к исключениям (InvalidOperationException) или повреждению данных.

Пример: проблема обычного Dictionary (повторение)

using System.Collections.Generic;
using System.Threading.Tasks;

Dictionary<int, int> concurrentDictProblem = new Dictionary<int, int>();

void AddToDict(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        // Попытка одновременного добавления/изменения
        // Приведет к исключениям или некорректному поведению
        concurrentDictProblem[start + i] = start + i;
    }
}

//Запуск примера в Main:
try
{
    Task t1 = Task.Run(() => AddToDict(0, 10000));
    Task t2 = Task.Run(() => AddToDict(5000, 10000)); // Пересечение ключей
    Task.WaitAll(t1, t2);
    Console.WriteLine($"Элементов в словаре (проблемном): {concurrentDictProblem.Count}");
}
catch (Exception ex)
{
    Console.WriteLine($"Ошибка в обычном словаре: {ex.Message}");
}

Этот код почти гарантированно выдаст исключение или зависнет из-за проблем с потокобезопасностью.

4. Основные операции

ConcurrentDictionary предоставляет специализированные атомарные операции «проверка + действие».

TryAdd(TKey key, TValue value): атомарно добавляет пару ключ-значение. Возвращает true, если ключ был добавлен, и false, если ключ уже существует.

ConcurrentDictionary<string, int> scores = new ConcurrentDictionary<string, int>();
if (scores.TryAdd("Alice", 100))
    Console.WriteLine("Alice добавлена."); // Вывод: Alice добавлена.
if (!scores.TryAdd("Alice", 150))
    Console.WriteLine("Alice уже есть."); // Вывод: Alice уже есть.

TryGetValue(TKey key, out TValue value): атомарно получает значение по ключу.

if (scores.TryGetValue("Alice", out int aliceScore))
    Console.WriteLine($"Счет Alice: {aliceScore}"); // Вывод: Счет Alice: 100

TryUpdate(TKey key, TValue newValue, TValue comparisonValue): атомарно обновляет значение, только если текущее равно comparisonValue. Предотвращает гонки.

// Текущее значение Alice = 100
if (scores.TryUpdate("Alice", 120, 100)) // Обновит 100 на 120
    Console.WriteLine("Счет Alice обновлен до 120."); // Вывод: Счет Alice обновлен до 120.
if (!scores.TryUpdate("Alice", 130, 100)) // Не обновит, т.к. текущее 120, а не 100
    Console.WriteLine("Счет Alice не обновлен (устаревшие данные)."); // Вывод: ...

TryRemove(TKey key, out TValue value): атомарно удаляет элемент по ключу.

if (scores.TryRemove("Alice", out int removedScore))
    Console.WriteLine($"Alice удалена, счет был: {removedScore}"); // Вывод: Alice удалена, счет был: 120

5. Продвинутые атомарные операции

Эти два метода — рабочие лошадки ConcurrentDictionary, покрывающие множество сценариев.

GetOrAdd(TKey key, TValue valueFactory(TKey key)): атомарно возвращает существующее значение по ключу либо создаёт и добавляет новое через фабрику. Идеально для кэшей и уникальных сущностей.

// Предположим, мы кэшируем тяжелые объекты
ConcurrentDictionary<int, HeavyObject> objectCache = new ConcurrentDictionary<int, HeavyObject>();

HeavyObject GetOrCreateHeavyObject(int id)
{
    // Если уже есть — вернёт его, иначе создаст и добавит
    return objectCache.GetOrAdd(id, (key) =>
    {
        Console.WriteLine($"Создаем новый HeavyObject для ID: {key}");
        return new HeavyObject(key); // Имитация создания дорогого объекта
    });
}

// В Main:
HeavyObject obj1 = GetOrCreateHeavyObject(1); // Создаст новый
HeavyObject obj2 = GetOrCreateHeavyObject(2); // Создаст новый
HeavyObject obj3 = GetOrCreateHeavyObject(1); // Вернет существующий obj1

AddOrUpdate(TKey key, TValue addValue, Func<TKey, TValue, TValue> updateValueFactory): атомарно добавляет значение, если ключ отсутствует, или обновляет существующее через фабрику.

• addValue: значение для добавления, если ключ не найден.
• updateValueFactory: функция, вычисляющая новое значение на основе ключа и текущего значения.

// Подсчет количества посещений страницы
ConcurrentDictionary<string, int> pageViews = new ConcurrentDictionary<string, int>();

void IncrementPageView(string page)
{
    pageViews.AddOrUpdate(page, 1, // Если страница новая, добавить 1
                          (key, existingVal) => existingVal + 1); // Иначе увеличить на 1
    Console.WriteLine($"Страница '{page}' посещена {pageViews[page]} раз.");
}

// В Main:
IncrementPageView("Home");   // Home: 1
IncrementPageView("About");  // About: 1
IncrementPageView("Home");   // Home: 2
IncrementPageView("Home");   // Home: 3
IncrementPageView("Contact"); // Contact: 1

6. Примеры использования для кэширования или управления состояниями

Кэширование данных: ConcurrentDictionary — отличный выбор для in-memory кэша: GetOrAdd предотвращает повторное создание дорогих объектов.

Управление пользовательскими сессиями: безопасное хранение и обновление данных сессий из разных запросов.

Подсчёт статистики: с помощью AddOrUpdate удобно инкрементировать счётчики событий, просмотров, голосов и т.д.

Реестры/Service Locator: хранение зарегистрированных сервисов или плагинов, доступных из разных потоков.

ConcurrentDictionary<TKey, TValue> — высокооптимизированная коллекция, существенно упрощающая многопоточную разработку со словарями за счёт набора атомарных операций без ручной синхронизации.