Developer

Потокобезопасные коллекции и Channels

↑ 2.3 Асинхронность и многопоточность · ← Предыдущая · → Следующая

Обязательно~4 мин чтенияУровень: middle

Зачем это на собесе

Спрашивают, какую коллекцию выбрать для очереди задач или кэша и чем Channel лучше BlockingCollection.

Объяснение

System.Collections.Concurrent

ТипНазначениеЗамечание
ConcurrentDictionary<K,V>кэш, реестрGetOrAdd/AddOrUpdate атомарны, но фабрики могут выполниться несколько раз
ConcurrentQueue<T>FIFO, lock-freeнет блокирующего ожидания
ConcurrentStack<T>LIFO
ConcurrentBag<T>неупорядоченный пул, оптимален когда продюсер = консюмерсложнее в отладке
BlockingCollection<T>блокирующая очередь producer/consumerблокирует потоки, не async
ImmutableArray/ImmutableDictionaryкопирование при записи, безопасное чтениеподходит для конфигураций

Channel<T>

Асинхронная очередь producer/consumer из System.Threading.Channels.

var channel = Channel.CreateBounded<Job>(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.Wait,
    SingleReader = false,
    SingleWriter = false,
});
 
// Продюсер
async Task ProduceAsync(IEnumerable<Job> jobs, CancellationToken ct)
{
    foreach (var j in jobs)
        await channel.Writer.WriteAsync(j, ct); // ждёт, если канал полон
    channel.Writer.Complete();
}
 
// Консюмеры
async Task ConsumeAsync(CancellationToken ct)
{
    await foreach (var job in channel.Reader.ReadAllAsync(ct))
        await job.RunAsync(ct);
}
 
await Task.WhenAll(
    ProduceAsync(source, ct),
    ConsumeAsync(ct), ConsumeAsync(ct), ConsumeAsync(ct));

Ограниченный (bounded) канал даёт backpressure: медленные потребители тормозят производителя, память не растёт бесконечно. Unbounded канал безопаснее только если поток данных заведомо мал.

Нюансы и подводные камни

  • ConcurrentDictionary потокобезопасен для отдельных операций, но составные («прочитал — проверил — записал») не атомарны.
  • Перечисление ConcurrentDictionary даёт снимок без гарантии согласованности.
  • Не забывайте Writer.Complete(), иначе читатели будут ждать вечно.
  • В Channel выбирайте FullMode осознанно: Wait, DropOldest, DropNewest, DropWrite.
  • SingleReader/SingleWriter = true включает более быстрые реализации, но нарушение контракта приводит к ошибкам.

Практика

  1. Построить конвейер «читаем файлы → парсим → пишем в БД» на трёх каналах.
  2. Заменить BlockingCollection в фоновом воркере на Channel и сравнить использование потоков.
  3. Реализовать кэш на ConcurrentDictionary<string, Lazy<Task<T>>>, чтобы дорогая загрузка выполнялась один раз.

Вопросы с ответами

Связанные темы

Вид графа

Обратные ссылки