Round #304

Implement async producer-consumer pattern using Channel<T>

**Async Producer-Consumer Pattern using Channel<T>** The producer-consumer pattern is a classic design pattern that allows for asynchronous communication between two or more threads. In this implementation, we will use the `Channel<T>` class in C# to create an async producer-consumer pattern. ### Overview * We will create a `Producer` class that produces items and writes them to a channel. * We will create a `Consumer` class that reads items from the channel and processes them. * We will use the `Channel<T>` class to create a channel that allows for asynchronous communication between the producer and consumer. ### Implementation ```csharp using System; using System.Threading; using System.Threading.Tasks; using System.Threading.Channels; public class Producer { private readonly Channel<int> _channel; public Producer(Channel<int> channel) { _channel = channel; } public async Task ProduceAsync(int count) { for (int i = 0; i < count; i++) { await _channel.Writer.WriteAsync(i); Console.WriteLine($"Produced: {i}"); await Task.Delay(100); } _channel.Writer.Complete(); } } public class Consumer { private readonly Channel<int> _channel; public Consumer(Channel<int> channel) { _channel = channel; } public async Task ConsumeAsync() { while (await _channel.Reader.WaitToReadAsync()) { while (_channel.Reader.TryRead(out int item)) { Console.WriteLine($"Consumed: {item}"); } } } } class Program { static async Task Main(string[] args) { var channel = Channel.CreateUnbounded<int>(); var producer = new Producer(channel); var consumer = new Consumer(channel); _ = producer.ProduceAsync(10); await consumer.ConsumeAsync(); Console.ReadLine(); } } ``` ### Explanation * We create a `Channel<int>` instance in the `Main` method and pass it to the `Producer` and `Consumer` classes. * The `Producer` class writes items to the channel using the `WriteAsync` method. * The `Consumer` class reads items from the channel using the `ReadAsync` method. * The `WaitToReadAsync` method is used to wait for items to be available in the channel. * The `TryRead` method is used to read items from the channel without blocking. * The `Complete` method is used to signal that no more items will be written to the channel. ### Example Use Cases * This implementation can be used in scenarios where data needs to be processed in real-time, such as in a streaming application. * It can also be used in scenarios where data needs to be processed in batches, such as in a data processing pipeline. ### Advice * Use the `Channel<T>` class to create a channel that allows for asynchronous communication between threads. * Use the `WriteAsync` method to write items to the channel. * Use the `ReadAsync` method to read items from the channel. * Use the `WaitToReadAsync` method to wait for items to be available in the channel. * Use the `TryRead` method to read items from the channel without blocking. * Use the `Complete` method to signal that no more items will be written to the channel.

```csharp using System; using System.Threading; using System.Threading.Tasks; using System.Threading.Channels; public class ProducerConsumerExample { public async Task RunExampleAsync() { // Create a bounded channel with a capacity of 10 items var channel = Channel.CreateBounded<int>(10); // Start the producer _ = ProduceNumbersAsync(channel.Writer, 1, 10); // Start the consumer await ConsumeNumbersAsync(channel.Reader); } private async Task ProduceNumbersAsync(ChannelWriter<int> writer, int start, int end) { try { for (int i = start; i <= end; i++) { // Write to the channel await writer.WriteAsync(i); Console.WriteLine($"Produced: {i}"); await Task.Delay(100); // Simulate production delay } // Complete the writer to signal the end of production writer.Complete(); } catch (Exception ex) { // Handle any exceptions that occur during production Console.WriteLine($"Production error: {ex.Message}"); writer.Complete(ex); } } private async Task ConsumeNumbersAsync(ChannelReader<int> reader) { try { while (await reader.WaitToReadAsync()) { while (reader.TryRead(out int item)) { // Process the item Console.WriteLine($"Consumed: {item}"); await Task.Delay(100); // Simulate consumption delay } } } catch (OperationCanceledException) { // Handle cancellation Console.WriteLine("Consumption cancelled"); } catch (Exception ex) { // Handle any other exceptions that occur during consumption Console.WriteLine($"Consumption error: {ex.Message}"); } } } class Program { static async Task Main(string[] args) { var example = new ProducerConsumerExample(); await example.RunExampleAsync(); } } ``` This implementation demonstrates the async producer-consumer pattern using `Channel<T>`. The `ProduceNumbersAsync` method writes numbers to the channel, while the `ConsumeNumbersAsync` method reads and processes the numbers from the channel. The `RunExampleAsync` method sets up the channel and starts both the producer and consumer. The `Channel.CreateBounded` method is used to create a bounded channel with a capacity of 10 items. This means that the producer will block if the channel is full, and the consumer will block if the channel is empty. The `ProduceNumbersAsync` method uses a `try`-`catch` block to handle any exceptions that occur during production. If an exception occurs, the writer is completed with the exception to propagate it to the consumer. The `ConsumeNumbersAsync` method uses a `try`-`catch` block to handle any exceptions that occur during consumption. If an exception occurs, it is logged to the console. The `Main` method in the `Program` class creates an instance of the `ProducerConsumerExample` class and runs the example asynchronously. Note that this implementation uses `await Task.Delay` to simulate production and consumption delays. In a real-world scenario, you would replace these with your actual production and consumption logic.