Unleash the power of Channel class for Pub/Sub in C#

The Channel class in C# is a robust tool for managing asynchronous communication between producers and consumers. Introduced with .NET Core 3.0 and now part of the core framework in .NET 7, the Channel class eliminates the need for additional NuGet packages. This blog post explores how to utilize Channel for efficient pub/sub design, covering the creation and usage of bounded and unbounded channels.

What is the Channel Class?

The Channel class provides a thread-safe data structure for asynchronously passing data between a producer and a consumer. It supports two main types of channels:

Unbounded Channels: Allow unlimited data with concurrent readers and writers.
Bounded Channels: Have a maximum capacity to control memory usage and flow.

Creating a Channel

Unbounded Channel

To create an unbounded channel, use the Channel.CreateUnbounded method. For example:

var channel = Channel.CreateUnbounded<string>(new UnboundedChannelOptions
{
    AllowSynchronousContinuations = false, // Default is false
    SingleReader = false, // Default is false
    SingleWriter = false  // Default is false
});

Here, the UnboundedChannelOptions allow customization of properties like:

AllowSynchronousContinuations: Enables synchronous continuations for improved throughput but may reduce parallelism.
SingleReader and SingleWriter: Ensure only one reader or writer operates at a time.

Bounded Channel

A bounded channel is created using Channel.CreateBounded with a specified capacity:

var channel = Channel.CreateBounded<string>(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.DropOldest // Drop the oldest item when full
});

The FullMode property defines the behavior when the channel is full:

Wait: Waits for space to become available.
DropWrite: Drops the item being written.
DropOldest: Removes the oldest item.
DropNewest: Removes the newest item.

Producing and Consuming Data

Channels use a Writer to enqueue data and a Reader to dequeue data. Here’s a simple example:

Writing Data (Producer)

var writer = channel.Writer;
for (int i = 0; i < 10; i++)
{
    await writer.WriteAsync($"Item {i}");
}
writer.Complete();

Reading Data (Consumer)

var reader = channel.Reader;
while (await reader.WaitToReadAsync())
{
    while (reader.TryRead(out var item))
    {
        Console.WriteLine(item);
    }
}

Decoupling Producers and Consumers

One of the most significant advantages of using the Channel class is the decoupling it provides between producers and consumers. Unlike traditional message queues or streams that operate out-of-process, Channel facilitates in-process communication, making it lightweight and efficient for many scenarios.

End-to-End Example

Here’s a complete example demonstrating a producer-consumer scenario:

var channel = Channel.CreateBounded<string>(new BoundedChannelOptions(10)
{
    FullMode = BoundedChannelFullMode.DropOldest
});

// Consumer Task
var consumerTask = Task.Run(async () =>
{
    var reader = channel.Reader;
    while (await reader.WaitToReadAsync())
    {
        while (reader.TryRead(out var item))
        {
            Console.WriteLine($"Consumed: {item}");
        }
    }
});

// Producer Task
var producerTask = Task.Run(async () =>
{
    var writer = channel.Writer;
    for (int i = 0; i < 20; i++)
    {
        await writer.WriteAsync($"Item {i}");
        Console.WriteLine($"Produced: Item {i}");
    }
    writer.Complete();
});

await Task.WhenAll(consumerTask, producerTask);

Conclusion

The Channel class is a powerful addition to the .NET ecosystem, offering an easy-to-use and highly efficient mechanism for producer-consumer communication. Whether you’re implementing a simple in-memory queue or building a complex pub/sub system, Channel provides the tools you need to get the job done. Experiment with bounded and unbounded channels to find the best fit for your application’s needs.

The entire exercise is available here on YouTube: https://youtu.be/NWRGEz-P2PU?si=pNOybA4ZVyyG_K8A