# Concurrent Collections

Course: [C# Advanced](https://www.devtutorials.ai/learn/advanced-csharp) · Section: Multithreading
Source: https://www.devtutorials.ai/learn/advanced-csharp/multithreading-parallel/concurrent-collections

> Share data between threads without locks: ConcurrentDictionary, ConcurrentQueue, ConcurrentStack, ConcurrentBag, producer-consumer, caching and data processing.

## Introduction to Concurrent Programming

Think about a busy restaurant kitchen with one order rail.

Waiters clip new orders onto the rail. Cooks take orders off it. Many people use the rail at the same time, and nobody has to stand guard over it. The rail itself is built so that two people never grab the same ticket.

In lesson 5 you protected a shared `List` with a `lock`. That works, but you must remember the lock **everywhere** the list is used. Forget it in one place, and you have a bug.

**Concurrent collections** are like the order rail: collections that are **built** to be safe when many threads use them at the same time. The safety is built into the collection, so your code stays simple and safe.

**In simple words:** concurrent programming means multiple pieces of work happening at the same time. Concurrent collections are the shared containers that let these pieces of work safely share data together.

They all live in the `System.Collections.Concurrent` namespace.

**Easy Way to Remember:** a normal collection needs **you** to add locks. A concurrent collection **locks itself**.

## What You Will Learn

1. `ConcurrentDictionary<TKey,TValue>`, the most useful one.
2. `ConcurrentQueue<T>` (first in, first out).
3. `ConcurrentStack<T>` (last in, first out).
4. `ConcurrentBag<T>` (no order).
5. The producer-consumer pattern.
6. What makes a collection thread-safe, and its limits.
7. Concurrent caching and concurrent data processing.

**About the Run buttons in this lesson.** The browser playground has only one thread, so the workers in these examples take turns instead of running at the same time. The collections behave exactly the same way, and the results are the same as on a real computer with multiple cores. Examples that pause a thread while waiting are shown as plain code without a Run button.

## Start With a Working Example

Press **Run**.

```csharp runnable title="Count words from many workers, no lock needed"
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading.Tasks;

string[] lines =
{
    "the cat sat on the mat",
    "the dog sat on the log",
    "a cat and a dog",
};

var counts = new ConcurrentDictionary<string, int>();

await Task.WhenAll(lines.Select(line => Task.Run(() =>
{
    foreach (string word in line.Split(' '))
        counts.AddOrUpdate(word, 1, (_, old) => old + 1);
})));

foreach (var pair in counts.OrderByDescending(p => p.Value).ThenBy(p => p.Key).Take(4))
    Console.WriteLine($"{pair.Key}: {pair.Value}");
```

Three workers update the same dictionary at the same time. `AddOrUpdate` adds a word with count 1, or adds one to its existing count, as one safe operation.

### Try This

Add a fourth line, `"the end"`, to the array. The count for `the` goes up to 5.

## ConcurrentDictionary&lt;TKey,TValue&gt;

**`ConcurrentDictionary<TKey,TValue>`** is a thread-safe dictionary. It has the usual indexer, plus methods that **check and change in one atomic step**. This is exactly what plain dictionaries cannot do safely when many threads run at once.

| Method | What it does, as one safe atomic step |
|---|---|
| `TryAdd(key, value)` | Adds the key only if it is not already there. Returns `true` if it added successfully |
| `TryGetValue(key, out value)` | Reads and returns a value if the key exists |
| `TryRemove(key, out value)` | Removes a key and returns its value |
| `TryUpdate(key, newValue, expected)` | Changes the value only if it still has the value you expected |
| `GetOrAdd(key, factory)` | Returns the existing value, or creates a new one and stores it |
| `AddOrUpdate(key, addValue, updateFactory)` | Adds a new value if the key is missing, or updates the existing value |
| `dict[key] = value` | Sets a value (adds a new key or replaces an existing one) |

```csharp runnable title="The main ConcurrentDictionary methods"
using System;
using System.Collections.Concurrent;

var stock = new ConcurrentDictionary<string, int>();

Console.WriteLine($"TryAdd pens:  {stock.TryAdd("pens", 40)}");
Console.WriteLine($"TryAdd pens again: {stock.TryAdd("pens", 99)}");

int books = stock.GetOrAdd("books", _ => 10);
Console.WriteLine($"GetOrAdd books: {books}");

int pens = stock.AddOrUpdate("pens", 1, (_, old) => old + 5);
Console.WriteLine($"AddOrUpdate pens: {pens}");

Console.WriteLine($"TryUpdate books 10 -> 8: {stock.TryUpdate("books", 8, 10)}");
Console.WriteLine($"TryUpdate books 10 -> 7: {stock.TryUpdate("books", 7, 10)}");

if (stock.TryRemove("pens", out int removed))
    Console.WriteLine($"Removed pens: {removed}");

Console.WriteLine($"Left: books = {stock["books"]}, count = {stock.Count}");
```

`TryUpdate` changes the value only if it matches the value you expected. The second call fails because `books` is already 8, not 10. This protects you: if another thread changed the value between when you read it and when you try to update it, your update fails instead of overwriting that change.

In `GetOrAdd` and `AddOrUpdate`, the factory (the code that creates a value) can run **more than once** when two threads both need to add the same key at the exact same time. Only one result is kept in the dictionary. This is why the factory code should be simple: do not put side effects inside the factory, such as sending emails, writing to a log, or updating a database. If the factory runs twice, you do not want the side effect to happen twice.

## ConcurrentQueue

**`ConcurrentQueue<T>`** is a thread-safe **first in, first out** (FIFO) queue, like a line at a ticket counter: the first person to arrive is the first person served.

| Method | What it does |
|---|---|
| `Enqueue(item)` | Adds an item to the back of the queue |
| `TryDequeue(out item)` | Takes and removes an item from the front. Returns `false` if the queue is empty |
| `TryPeek(out item)` | Looks at the item at the front without removing it |
| `IsEmpty`, `Count` | Checks if the queue is empty, or gets the number of items |

There is no plain `Dequeue()` method, only `TryDequeue()`. This is because between the moment you check if the queue has items and the moment you take one, another thread might remove that item first. `TryDequeue` tells you whether **you** successfully got an item.

```csharp runnable title="Print jobs in the order they arrived"
using System;
using System.Collections.Concurrent;

var printQueue = new ConcurrentQueue<string>();
printQueue.Enqueue("invoice.pdf");
printQueue.Enqueue("photo.jpg");
printQueue.Enqueue("letter.docx");

if (printQueue.TryPeek(out string? next))
    Console.WriteLine($"Next up: {next}");

while (printQueue.TryDequeue(out string? job))
    Console.WriteLine($"Printing {job}");

Console.WriteLine($"Queue empty: {printQueue.IsEmpty}");
```

## ConcurrentStack

**`ConcurrentStack<T>`** is a thread-safe **last in, first out** (LIFO) stack, like a stack of plates: you take the plate you put down last.

| Method | What it does |
|---|---|
| `Push(item)` / `PushRange(items)` | Adds one or more items to the top |
| `TryPop(out item)` | Takes and removes the item from the top |
| `TryPopRange(array)` | Takes several items from the top at once |
| `TryPeek(out item)` | Looks at the top item without removing it |

```csharp runnable title="Undo the most recent action first"
using System;
using System.Collections.Concurrent;

var undo = new ConcurrentStack<string>();
undo.Push("typed 'Hello'");
undo.Push("made it bold");
undo.Push("changed the colour");

while (undo.TryPop(out string? action))
    Console.WriteLine($"Undo: {action}");
```

## ConcurrentBag

**`ConcurrentBag<T>`** is a thread-safe collection with **no guaranteed order**. Think of a bag of marbles: you put marbles in, and you take out whichever one you reach first.

`ConcurrentBag` is very fast when the same thread both adds and takes items. This is because each thread keeps its own separate pile inside the bag, and threads mainly work with their own pile.

| Method | What it does |
|---|---|
| `Add(item)` | Adds an item to the bag |
| `TryTake(out item)` | Takes and removes any item from the bag. Returns `false` if the bag is empty |
| `Count`, `IsEmpty` | Gets the number of items or checks if the bag is empty |

```csharp runnable title="Collect results from many workers"
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading.Tasks;

var results = new ConcurrentBag<int>();

await Task.WhenAll(Enumerable.Range(1, 10).Select(n => Task.Run(() =>
{
    if (n % 2 == 0) results.Add(n * n);
})));

Console.WriteLine($"Collected {results.Count} squares");
Console.WriteLine($"Sorted: {string.Join(", ", results.OrderBy(x => x))}");
```

The bag does not keep any specific order, so sort the results before displaying if order matters.

| Collection | Order | Typical use |
|---|---|---|
| `ConcurrentDictionary<K,V>` | By key | Counts, caches, lookups |
| `ConcurrentQueue<T>` | First in, first out | Job queues, messages |
| `ConcurrentStack<T>` | Last in, first out | Undo, depth-first work |
| `ConcurrentBag<T>` | None | Collecting results from workers |

## Producer-Consumer Pattern

In the **producer-consumer** pattern, one or more **producers** create work items, and one or more **consumers** process them. A thread-safe queue sits between them.

```text
Producer(s) --> [ queue ] --> Consumer(s)
```

It is the restaurant again: waiters (producers) add orders to the rail, cooks (consumers) take them off. The producers and consumers work independently. The producers can add orders faster than the cooks work, and the queue holds the waiting orders. You can add more cooks when too many orders pile up.

The modern way to build this in async code is a **channel** from `System.Threading.Channels`. The consumer waits for the next item without blocking a thread.

```csharp runnable title="A producer and two consumers with a channel"
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading.Channels;
using System.Threading.Tasks;

var channel = Channel.CreateUnbounded<int>();
var processed = new ConcurrentBag<string>();

Task producer = Task.Run(async () =>
{
    for (int order = 1; order <= 6; order++)
    {
        await channel.Writer.WriteAsync(order);
        await Task.Delay(20);
    }
    channel.Writer.Complete(); // no more orders
});

Task Consumer() => Task.Run(async () =>
{
    await foreach (int order in channel.Reader.ReadAllAsync())
        processed.Add($"order {order}");
});

await Task.WhenAll(producer, Consumer(), Consumer());

Console.WriteLine($"Orders cooked: {processed.Count}");
Console.WriteLine(string.Join(", ", processed.OrderBy(p => p.Length).ThenBy(p => p)));
```

`Writer.Complete()` signals to the consumers "I am done adding items, no more will come". When the channel is empty and `Complete()` has been called, the `await foreach` loops in the consumers end.

Older code uses **`BlockingCollection<T>`**, which pauses the consumer thread while it waits for items. It does not work in the browser's single-threaded playground, so it is shown as an example only:

```text
using var buffer = new BlockingCollection<int>(boundedCapacity: 10);

var producer = Task.Run(() =>
{
    for (int i = 1; i <= 100; i++) buffer.Add(i);   // pauses if the buffer is full
    buffer.CompleteAdding();
});

var consumer = Task.Run(() =>
{
    foreach (int item in buffer.GetConsumingEnumerable())   // waits for each item
        Process(item);
});

Task.WaitAll(producer, consumer);
```

A **bounded** channel or collection has a maximum size. When it is full, the producer pauses until a consumer takes an item. This prevents a fast producer from creating work items faster than the consumers can handle, which would fill up memory.

```text
var channel = Channel.CreateBounded<int>(capacity: 10);
```

## Thread-Safe Collections

A concurrent collection guarantees that **each single method call** is safe. It does **not** guarantee that **a sequence of multiple calls** is safe.

```text
// Not safe, even with a ConcurrentDictionary
if (!cache.ContainsKey(key))        // 1. check
    cache[key] = LoadValue(key);    // 2. act: another thread may have added it in between

// Safe: one call does both checks and acts
cache.GetOrAdd(key, LoadValue);
```

```text
// Not safe: Count may change before TryDequeue
if (queue.Count > 0)
    queue.TryDequeue(out var item);

// Safe: TryDequeue tells you if you got something
if (queue.TryDequeue(out var item)) { ... }
```

**Easy Way to Remember:** use the **Try** and **GetOr** methods. They check and act in one atomic step.

There are also read-only options that are safe because nothing can ever change:

| Type | Safe because |
|---|---|
| `ImmutableList<T>`, `ImmutableDictionary<K,V>` | Every change creates and returns a **new** collection; the original never changes |
| `FrozenDictionary<K,V>` (.NET 8+) | Built once, very fast to read, never changes |
| `IReadOnlyList<T>` over data nobody modifies | Nobody can write to it, so nothing to protect |

## Concurrent Caching

A **cache** stores results that are expensive to compute (slow, use a lot of CPU, or talk to a database). When another request needs the same thing, you get the answer instantly from the cache instead of computing it again. Many threads read the cache at once, so it must be thread-safe.

`GetOrAdd` is a simple one-line cache. To guarantee that the slow work runs **only once** per key, even when two threads ask for the same key at the exact same moment, store a **`Lazy<T>`** inside the dictionary:

```csharp runnable title="A cache where each value is computed once"
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

var cache = new ConcurrentDictionary<int, Lazy<string>>();
int loads = 0;

string GetProduct(int id) =>
    cache.GetOrAdd(id, key => new Lazy<string>(() =>
    {
        Interlocked.Increment(ref loads);   // the slow part
        return $"Product #{key}";
    })).Value;

int[] requests = { 1, 2, 1, 3, 2, 1, 1, 3, 2, 1 };
string[] answers = await Task.WhenAll(requests.Select(id => Task.Run(() => GetProduct(id))));

Console.WriteLine($"Requests: {answers.Length}");
Console.WriteLine($"Database loads: {loads}");
Console.WriteLine($"First answer: {answers[0]}");
```

Ten requests come in, but only three loads happen (one for each distinct product ID). Even if two threads create a `Lazy` for the same key at the same time, `GetOrAdd` keeps only one. `Lazy` runs its factory function only once, no matter how many threads ask for it.

In ASP.NET Core applications, use the built-in `IMemoryCache` instead. It is built on the same idea but adds expiry times (the cache forgets old items) and memory limits (the cache cannot grow bigger than a set size).

## Concurrent Data Processing

Pick the collection based on the **order** you need: by key (`ConcurrentDictionary`), first in first out (`ConcurrentQueue`), last in first out (`ConcurrentStack`), or no order at all (`ConcurrentBag`).

A real-world job: process a big batch of records using several workers, and collect totals as they work.

```csharp runnable title="Total sales by city from many workers"
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading.Tasks;

(string City, decimal Amount)[] sales =
{
    ("Chennai", 250), ("Delhi", 120), ("Chennai", 980), ("Mumbai", 45),
    ("Delhi", 300), ("Mumbai", 610), ("Chennai", 75), ("Delhi", 90),
};

var totals = new ConcurrentDictionary<string, decimal>();
var bigSales = new ConcurrentBag<decimal>();

Parallel.ForEach(sales, sale =>
{
    totals.AddOrUpdate(sale.City, sale.Amount, (_, old) => old + sale.Amount);
    if (sale.Amount >= 500) bigSales.Add(sale.Amount);
});

foreach (var city in totals.OrderBy(t => t.Key))
    Console.WriteLine($"{city.Key,-8} {city.Value,6}");

Console.WriteLine($"Sales of 500 or more: {string.Join(", ", bigSales.OrderBy(x => x))}");
```

`Parallel.ForEach` (lesson 7) divides the sales list among workers. The concurrent dictionary and bag let every worker add its results safely without any manual locks in your code.

## Common mistakes

| Mistake | Wrong | Correct | Why |
|---|---|---|---|
| Sharing a plain `Dictionary` between threads | `dict[word]++` from multiple tasks | `counts.AddOrUpdate(word, 1, (_, n) => n + 1)` | `Dictionary` is not thread-safe |
| Check in one call, act in another | `if (!d.ContainsKey(k)) d[k] = v;` | `d.GetOrAdd(k, v);` | Another thread can add the key in between your check and your action |
| Checking `Count` before taking an item | `if (q.Count > 0) q.TryDequeue(...)` | `if (q.TryDequeue(out var x))` | Another thread may take the last item between your check and your take |
| Running side effects in a factory | `GetOrAdd(k, _ => SendEmail())` | Keep the factory simple; if you need side effects, use `Lazy<T>` | The factory can run multiple times |
| Expecting a `ConcurrentBag` to preserve order | Printing a bag directly | `bag.OrderBy(x => x)` | A bag is optimised for speed, not order |
| Consumer loop never exits | Forgetting `Writer.Complete()` | Always call `Complete()` when the producer is done | Without it, `ReadAllAsync` waits forever for more items |
| Unbounded queue with a very fast producer | `Channel.CreateUnbounded<T>()` for massive input | `Channel.CreateBounded<T>(100)` | An unbounded queue can fill memory if producers are much faster than consumers |

## What to Remember

- Concurrent collections in `System.Collections.Concurrent` are safe to use from many threads without adding your own locks.
- `ConcurrentDictionary` is the most useful; `GetOrAdd`, `AddOrUpdate`, and `TryUpdate` check and change in one atomic step.
- `ConcurrentQueue` is first in, first out; `ConcurrentStack` is last in, first out; `ConcurrentBag` has no order.
- Use the `Try...` methods: an item can disappear between checking and taking, so let the method tell you if it succeeded.
- Producer-consumer puts a queue between the code that creates work and the code that processes it; use channels in async code.
- Each method call is safe, but a sequence of separate calls is not. Use single-step methods whenever possible.
- `ConcurrentDictionary` with `Lazy<T>` makes a cache that computes each value exactly once, even with many threads asking at once.
- Concurrent collections plus `Parallel.ForEach` let many workers collect results without writing any synchronization code yourself.

Next: Parallel Programming