using System;

using System.Collections.Generic;

using System.Collections.Concurrent;

using System.Threading;

using System.Threading.Tasks;

using System.Linq;

​

public class Program

{

    public static void Main()

    {

        const int degreeOfParallelism = 2;

        CancellationToken token = new CancellationTokenSource(3000).Token;

        ThreadPool.SetMinThreads(degreeOfParallelism, Environment.ProcessorCount);

        try

        {

            Partitioner.Create(Iterator(token), EnumerablePartitionerOptions.NoBuffering)

                .AsParallel()

                .WithDegreeOfParallelism(degreeOfParallelism)

                .WithCancellation(token)

                .ForAll(item => ProcessItemAsync(item).GetAwaiter().GetResult());

        }

        catch (OperationCanceledException) { } // Ignore

    }

​

    private static IEnumerable<Item> Iterator(CancellationToken token)

    {

        while (true)

        {

            Task delayTask = Task.Delay(500, token);

            Print($"Iterator before GetDatabaseItems");

            foreach (Item item in GetDatabaseItems())

                yield return item;

            delayTask.GetAwaiter().GetResult();

        }

    }

​

    private static int _idSeed = 0;

    private class Item { public int Id { get; } = Interlocked.Increment(ref _idSeed); }

​

    private static Item[] GetDatabaseItems() => new Item[] {new(), new(), new()};

    private static async Task ProcessItemAsync(Item item)

    {

        Print($"Processing item #{item.Id}");

        await Task.Delay(100);

    }

​

    private static void Print(object value)

    {

        Console.WriteLine($@"{DateTime.Now:HH:mm:ss.fff} [{Thread.CurrentThread

            .ManagedThreadId}] > {value}");

    }

}