Channels in C#: The Perfect In-Memory Queue

Today I want to talk about a type in C# that most of you probably don't know. In fact, I didn’t know about it myself until a couple of years ago, when I joined my current client. Explaining it now is also perfect since it relates to what has recently happened with the MediatR library as it can replace part of its functionality. 

In today’s post, we’ll take a look at the Channels type in C#.

1 - What are Channels in C#?

Within C#, there’s a very specific type based on the concept of sending messages or events through that channel, meaning one part of the system creates an event and another part receives it. These events can be of any type and stay stored in the channel until the receiver consumes them.

If you think about it, this is very similar logic to what is known as the in-memory producer-consumer pattern. 

NOTE: If you’re not familiar with the producer-consumer pattern, I recommend reading this post first: Link to Producer-Consumer Pattern.

But to sum up, it’s a pattern where one part of the system generates an event, and another part consumes it. Neither part has knowledge of the other, and technically (even if not ideal for every system) it doesn’t matter who produces or consumes the event; consumers don’t care who produced it and vice versa. If you’re used to distributed systems, it’s just like a message queue. 

Channels allow us to implement this functionality, which is basically an in-memory message queue like those found in distributed systems, without any additional infrastructure. This enables decoupling both sides of the system by not having direct communication, and processing can happen asynchronously.

2 - What does this have to do with MediatR?

Some of you might be wondering what this has to do with using MediatR, since as we’ve seen, channels aren’t related to the mediator pattern. And that’s true, but the MediatR library doesn’t just implement the mediator pattern; it also implements notifications using its INotificationHandler<T>, and that’s where channels come in.

For this example, let’s take the same scenario as in the mediator pattern post, where an API endpoint receives a call to update a product. This endpoint invokes a handler with the logic and ends up creating a notification through MediatR. This notification is then listened to by another part of the system which sends an email to all interested users. 

If you read that post, or the one I made about Core-Driven Architecture, you’ll know I’m not a big fan of using MediatR, but I do find notifications very useful for decoupling the different parts of a system. 

So, let’s look at how to remove MediatR from our system and completely replace it with features already available in .NET. 

For the mediator pattern scenario, meaning the handlers, we simply remove them and make a direct call. You might think this couples the code, and in a way it does, but the code is logically coupled anyway, so the difference is almost zero.

If you use the MediatR pipeline with different behaviors, you can always replace them with middlewares and filters; it works basically the same way.

For notifications, we’ll use channels, and in the next section, we'll see how to configure it. 

This results in the following diagram: 

3 - Implementing the Producer-Consumer Pattern with Channels

As always, the code for this example is on GitHub, but my goal is to keep it simple.

First, let me mention we have an input DTO called ItemDto

public record ItemDto(int Id, decimal Price, string Title);

NOTE: In this example, for simplicity, I’m inserting it into the data layer, but in a real application it should be an entity, since they are different types. 

A very simple endpoint that just calls the use case to update the product: 

app.MapPut("items", async (UpdateItem updateItem, ItemDto itemDto) =>{    return await updateItem.Execute(itemDto);});

And for now, the use case where we read from the database and update the value:

public class UpdateItem (IDatabaseRepository databaseRepository){    public async Task<bool> Execute(ItemDto itemToUpdate)    {        if (itemToUpdate.Title.Length > 200)            throw new Exception("Title must be less than 200 characters");        if (itemToUpdate.Price <= 0)            throw new Exception("It can't be free");        await databaseRepository.UpdateItem(itemToUpdate.Id, itemToUpdate.Price, itemToUpdate.Title);                  return true;    }}

Note: The database layer is not implemented, it's just for demonstration purposes.

3.1 - Generating Events with Channels 

Let's suppose every time a product is updated and its price is 30% lower than before, we want to generate a notification—maybe an email to customers, like Steam does, a push notification, or anything else—it doesn't matter. The point is, we need to generate a notification and obviously, we want it to happen asynchronously. We don’t want to wait for that process to finish just to handle it ourselves. 

For this, in the MediatR post, we saw we could use the INotification type, although there are other solutions, like using an app such as hangfire. 

But in our current scenario, what we have to do is to import using System.Threading.Channels and add the channel in our use case’s constructor and use it to write into the channel: 

public class UpdateItem (IDatabaseRepository databaseRepository, Channel<ItemUpdated> channel 👈){    public async Task<bool> Execute(ItemDto itemToUpdate)    {        if (itemToUpdate.Title.Length > 200)            throw new Exception("Title must be less than 200 characters");        if (itemToUpdate.Price <= 0)            throw new Exception("It can't be free");        ItemDto originalItem = await databaseRepository.GetItemById(itemToUpdate.Id);        await databaseRepository.UpdateItem(itemToUpdate.Id, itemToUpdate.Price, itemToUpdate.Title);        decimal percentageDifference = ((itemToUpdate.Price - originalItem.Price) / originalItem.Price) * 100;        if (percentageDifference <= -30)        {            await channel.Writer.WriteAsync(new ItemUpdated() 👈            {                Id = itemToUpdate.Id,                NewPrice = itemToUpdate.Price,                NewTitle = itemToUpdate.Title,                OldPrice = originalItem.Price,                OldTitle = originalItem.Title            });        }        return true;    }}

And don’t forget to add it to the dependency container; here it’s not just about adding it, you have two options: create bounded or unbounded channels.

The difference is simple: bounded lets you set a maximum number of elements in the channel, while unbounded has no limit—well, technically yes, the system memory where it's running.  

Generally, I configure them as unbounded, but if you set them with limits, you also need to configure what to do when the channel fills up. Here are your options:

• Wait: Wait until there is space.
• DropWrite: Do not write; the message you’re trying to write is discarded.
• DropOldest: Remove the oldest message and add the new one.
• DropNewest: Remove the most recent message and add the new one.

Depending on what kind of system you’re building, some options may suit you more than others, or you might just want it without a limit if memory isn’t a concern—which nowadays is typically the case. 

This would be the implementation for a Channel with a capacity of 100 elements where we wait for previous events to free up space:

BoundedChannelOptions options = new BoundedChannelOptions(100) {    FullMode = BoundedChannelFullMode.Wait};builder.Services.AddSingleton(Channel.CreateBounded<ItemUpdated>(options));

3.2 - Consuming Events with Channels

Finally, all that’s left is to consume the event. You do this with a background task, in other words, a native .NET BackgroundService

public class ItemUpdatedWorker(Channel<ItemUpdated> channel) : BackgroundService{    protected override async Task ExecuteAsync(CancellationToken stoppingToken)    {        while (await channel.Reader.WaitToReadAsync(stoppingToken))        {            while (channel.Reader.TryRead(out ItemUpdated? item))            {                Console.WriteLine($"Item {item.Id} has been updated. Old price: {item.OldPrice}, " +                                  $"New price: {item.NewPrice}. Old title: {item.OldTitle}, " +                                  $"New title: {item.NewTitle}");            }        }    }}

And don’t forget to add it to the dependency container:

builder.Services.AddHostedService<ItemUpdatedWorker>();

As you can see, we just read from the channel. In this example, I’m printing the values to the console, but this background worker could perform any necessary action in the background.