namespace AppParallel
{
    internal class Program
    {
        static object lockObject = 
        new object();
        static void Main(string[] args)
        {
            // 創(chuàng)建示例數(shù)據(jù)
            var largeList = 
            Enumerable.Range(1, 1000000).ToList();

            // 設(shè)置并行選項(xiàng)
            var parallelOptions = new ParallelOptions
            {
                MaxDegreeOfParallelism = 
                Environment.ProcessorCount 
                // 使用處理器核心數(shù)量的線程
            };

            try
            {
                Parallel.ForEach(largeList, parallelOptions,
                (number) =>
                {
                    // 這里是對(duì)每個(gè)元素的處理邏輯
                    var result = ComplexCalculation(number);

                    // 注意：如果需要收集結(jié)果，要考慮線程安全
                    lock (lockObject)
                    {
                        // 進(jìn)行線程安全的結(jié)果收集
                        Console.WriteLine(result);
                    }
                });
            }
            catch (AggregateException ae)
            {
                // 處理并行處理中的異常
                foreach (var ex in 
                ae.InnerExceptions)
                {
                    Console.WriteLine($"Error: 
                    {ex.Message}");
                }
            }
        }
        private static int
        ComplexCalculation(int number)
        {
            // 模擬復(fù)雜計(jì)算
            Thread.Sleep(100);
            return number * 2;
        }

    }
}

namespace AppParallel
{
    internal class Program
    {
        static void Main(string[] args)
        {
            var largeList = Enumerable.Range(1, 1000000).ToList();
            ProcessByChunks(largeList, 1000); 
            // 每1000個(gè)元素一個(gè)塊
        }
        public static void ProcessByChunks<T>(List<T> largeList, 
        int chunkSize)
        {
            // 計(jì)算需要多少個(gè)分塊
            int chunksCount = (int)Math.Ceiling((double)largeList.Count / chunkSize);
            var tasks = new List<Task>();

            for (int i = 0; i < chunksCount; i++)
            {
                // 獲取當(dāng)前分塊的數(shù)據(jù)
                var chunk = largeList
                    .Skip(i * chunkSize)
                    .Take(chunkSize)
                    .ToList();

                // 創(chuàng)建新任務(wù)處理當(dāng)前分塊
                var task = Task.Run(() => ProcessChunk(chunk));
                tasks.Add(task);
            }

            // 等待所有任務(wù)完成
            Task.WaitAll(tasks.ToArray());
        }

        private static void 
        ProcessChunk<T>(List<T> chunk)
        {
            foreach (var item in chunk)
            {
                // 處理每個(gè)元素
                ProcessItem(item);
            }
        }

        private static void 
        ProcessItem<T>(T item)
        {
            // 具體的處理邏輯
            Console.WriteLine
            ($"Processing item: {item} on thread: {Task.CurrentId}");
        }

    }
}

public class ProducerConsumerExample
{
    private readonly BlockingCollection<int> _queue;
    private readonly 
    int _producerCount;
    private readonly 
    int _consumerCount;
    private readonly 
    CancellationTokenSource _cts;

    public ProducerConsumerExample(int queueCapacity = 1000)
    {
        _queue = new BlockingCollection<int>(queueCapacity);
        _producerCount = 1;
        _consumerCount = 
        Environment.ProcessorCount;
        _cts = new CancellationTokenSource();
    }

    public async Task ProcessDataAsync(List<int> largeList)
    {
        // 創(chuàng)建生產(chǎn)者任務(wù)
        var producerTask = 
        Task.Run(() => Producer(largeList));

        // 創(chuàng)建消費(fèi)者任務(wù)
        var consumerTasks = Enumerable.Range(0, _consumerCount)
            .Select(_ => Task.Run(() => Consumer()))
            .ToList();

        // 等待所有生產(chǎn)者完成
        await producerTask;

        // 標(biāo)記隊(duì)列已完成
        _queue.CompleteAdding();

        // 等待所有消費(fèi)者完成
        await Task.WhenAll(consumerTasks);
    }

    private void Producer(List<int> items)
    {
        try
        {
            foreach (var item in items)
            {
                if (_cts.
                Token.IsCancellationRequested)
                    break;

                _queue.Add(item);
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Producer error: 
            {ex.Message}");
            _cts.Cancel();
        }
    }

    private void Consumer()
    {
        try
        {
            foreach (var item in _queue.GetConsumingEnumerable())
            {
                if (_cts.Token.IsCancellationRequested)
                    break;

                // 處理數(shù)據(jù)
                ProcessItem(item);
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Consumer error: {ex.Message}");
            _cts.Cancel();
        }
    }

    private void ProcessItem(int item)
    {
        // 具體的處理邏輯
        Thread.Sleep(100);
        // 模擬耗時(shí)操作
        Console.WriteLine($"Processed item {item} on thread {Task.CurrentId}");
    }
}

// 使用示例
static async Task Main(string[] args)
{
    var processor = new ProducerConsumerExample();
    var largeList = Enumerable.Range(1, 10000).ToList();
    await processor.ProcessDataAsync(largeList);
}