Source High-Throughput Producer-Consumer with System.Threading.Channels in .NET - UnreliableCode Forums

AsyncMaster

Concurrency Geek

MEMBER

Rep: 72

Join Date: Jul 2020

Posts: 13

Thanks: 83

3y ago · Jun 18, 2023 3:40 PM

#1

If you are still using

CODE

BlockingCollection<T>

or raw

CODE

ConcurrentQueue<T>

with manual locks in modern .NET, you should switch to System.Threading.Channels!

Channels are designed for asynchronous producer-consumer pipelines with zero thread blocking and support for backpressure:

CSHARP

using System.Threading.Channels;

public class LogProcessingPipeline
{
    // Bounded channel: Holds up to 10,000 items. Drops oldest or blocks producer when full
    private readonly Channel<LogMessage> _channel = Channel.CreateBounded<LogMessage>(new BoundedChannelOptions(10_000)
    {
        SingleWriter = false,
        SingleReader = true,
        FullMode = BoundedChannelFullMode.Wait
    });

    // Producer (Multiple threads can call this)
    public async ValueTask PublishLogAsync(LogMessage msg, CancellationToken ct = default)
    {
        await _channel.Writer.WriteAsync(msg, ct);
    }

    // Consumer (Single dedicated worker thread)
    public async Task StartConsumerAsync(CancellationToken ct)
    {
        ChannelReader<LogMessage> reader = _channel.Reader;
        
        while (await reader.WaitToReadAsync(ct))
        {
            while (reader.TryRead(out LogMessage item))
            {
                await WriteToDiskAsync(item);
            }
        }
    }
}

If you are still using [code]BlockingCollection<T>[/code] or raw [code]ConcurrentQueue<T>[/code] with manual locks in modern .NET, you should switch to [b]System.Threading.Channels[/b]!

Channels are designed for asynchronous producer-consumer pipelines with zero thread blocking and support for backpressure:

[code=csharp]using System.Threading.Channels;

public class LogProcessingPipeline
{
    // Bounded channel: Holds up to 10,000 items. Drops oldest or blocks producer when full
    private readonly Channel<LogMessage> _channel = Channel.CreateBounded<LogMessage>(new BoundedChannelOptions(10_000)
    {
        SingleWriter = false,
        SingleReader = true,
        FullMode = BoundedChannelFullMode.Wait
    });

// Producer (Multiple threads can call this)
    public async ValueTask PublishLogAsync(LogMessage msg, CancellationToken ct = default)
    {
        await _channel.Writer.WriteAsync(msg, ct);
    }

// Consumer (Single dedicated worker thread)
    public async Task StartConsumerAsync(CancellationToken ct)
    {
        ChannelReader<LogMessage> reader = _channel.Reader;
        
        while (await reader.WaitToReadAsync(ct))
        {
            while (reader.TryRead(out LogMessage item))
            {
                await WriteToDiskAsync(item);
            }
        }
    }
}[/code]

AsyncMaster · Concurrency Geek
Task Parallel Library (TPL), async/await internals, and lock...

The following users thanked AsyncMaster for this post:

ThreadRacer

High Performance C++

VIP

Rep: 71

Join Date: Apr 2023

Posts: 9

Thanks: 17

3y ago · Jun 18, 2023 5:11 PM

#2

CODE

SingleWriter = true

and

CODE

SingleReader = true

options are huge for performance! When you enable those flags, the Channel implementation switches from multi-producer lock-free queues to an optimized single-producer algorithm with almost zero CAS overhead.

ThreadRacer · High Performance C++
Lock-free SPSC/MPMC queues, atomic memory orders, and low-la...

DevDan

.NET Core & Cloud

MEMBER

Rep: 324

Join Date: Feb 2021

Posts: 16

Thanks: 65

3y ago · Jun 19, 2023 6:11 PM

#3

We replaced an old RabbitMQ memory queue with an in-memory Bounded Channel for telemetry batching and were able to push over 2.5 million items/sec on a single quad-core VM. Clean and robust!

DevDan · .NET Core & Cloud
Writing clean C# code and microservices since .NET Core 2.1....

UnreliableCode Community

High-Throughput Producer-Consumer with System.Threading.Channels in .NET

Create Forum Thread

Add Reputation