C# 并行编程 - Parallel 与 PLINQ

using System.Threading.Tasks;

// 基本用法
Parallel.For(0, 100, i => {
    Console.WriteLine($"索引 {i}，线程: {Thread.CurrentThread.ManagedThreadId}");
});

// 带并行选项
Parallel.For(0, 1000, new ParallelOptions 
{ 
    MaxDegreeOfParallelism = 4  // 最大并行度
}, i => {
    // 计算密集型工作
    double result = Math.Sqrt(i) * Math.PI;
});

// 使用线程局部变量减少锁竞争
long sum = 0;
Parallel.For(0, 1000000,

    // 初始化局部变量
    () => 0L,

    // 循环体
    (i, state, local) => {
        return local + i;
    },

    // 聚合局部结果
    local => Interlocked.Add(ref sum, local)
);

Console.WriteLine($"总和: {sum}");

var items = Enumerable.Range(1, 100).ToList();

Parallel.ForEach(items, item => {
    ProcessItem(item);
});

// 并行遍历数组
int[] numbers = new int[1000];
Parallel.ForEach(numbers, (num, state, index) => {
    numbers[index] = index * index;
});

// 控制并行行为
Parallel.ForEach(items, new ParallelOptions { MaxDegreeOfParallelism = 4 },
    item => {
        Console.WriteLine($"{item}");
    });

Parallel.For(0, 1000000, (i, state) => {
    if (i > 1000)
    {
        state.Break();  // 通知不再处理后续迭代，但已完成的不影响
        // state.Stop(); // 立即停止所有正在执行的迭代
        return;
    }
    DoWork(i);
});

Parallel.Invoke(
    () => DoWork1(),
    () => DoWork2(),
    () => DoWork3(),
    () => DownloadData()
);

// 带配置
Parallel.Invoke(new ParallelOptions { MaxDegreeOfParallelism = 2 },
    () => LongRunningTaskA(),
    () => LongRunningTaskB()
);

using System;
using System.Drawing;
using System.Threading.Tasks;

class ImageProcessor
{
    public static void ApplyGrayscale(Bitmap bitmap)
    {
        Rectangle rect = new Rectangle(0, 0, bitmap.Width, bitmap.Height);
        System.Drawing.Imaging.BitmapData bmpData = bitmap.LockBits(rect,
            System.Drawing.Imaging.ImageLockMode.ReadWrite,
            bitmap.PixelFormat);

        int bytesPerPixel = Image.GetPixelFormatSize(bitmap.PixelFormat) / 8;
        byte* ptr = (byte*)bmpData.Scan0;
        int stride = bmpData.Stride;

        // 并行处理每一行
        Parallel.For(0, bitmap.Height, y =>
        {
            byte* row = ptr + (y * stride);
            for (int x = 0; x < bitmap.Width; x++)
            {
                int idx = x * bytesPerPixel;
                byte gray = (byte)(row[idx] * 0.299 + row[idx + 1] * 0.587 + row[idx + 2] * 0.114);
                row[idx] = gray;
                row[idx + 1] = gray;
                row[idx + 2] = gray;
            }
        });

        bitmap.UnlockBits(bmpData);
    }
}

using System.Linq;

var numbers = Enumerable.Range(1, 10000000);

// 并行查询
var evenNumbers = numbers
    .AsParallel()
    .Where(n => n % 2 == 0)
    .ToList();

// 并行聚合
var sum = numbers.AsParallel().Sum();
var average = numbers.AsParallel().Average();

// 自定义聚合
var result = numbers.AsParallel()
    .Aggregate(
        0,
        (local, n) => local + n,
        (total, local) => total + local,
        total => total
    );

// AsOrdered 保持原始顺序（有性能开销）
var ordered = numbers.AsParallel().AsOrdered()
    .Where(x => x > 5000000)
    .ToList();

// AsUnordered 取消顺序约束，提升性能
var unordered = numbers.AsParallel().AsUnordered()
    .Where(x => x > 5000000)
    .ToList();

var result = numbers.AsParallel()
    .WithDegreeOfParallelism(4)
    .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
    .Where(n => IsPrime(n))
    .ToList();

// 强制顺序执行（调试使用）
var sequential = numbers.AsParallel()
    .WithExecutionMode(ParallelExecutionMode.Default)
    .AsSequential()  // 切换回顺序 LINQ
    .Where(n => n % 2 == 0);

// 控制结果返回方式
var results = numbers.AsParallel()
    .WithMergeOptions(ParallelMergeOptions.NotBuffered)  // 立即返回
    .Select(x => HeavyCompute(x))
    .ToList();

// 合并选项说明：
// - NotBuffered: 结果一产生就返回
// - AutoBuffered: 系统决定缓冲大小（默认）
// - FullyBuffered: 全部完成后再返回

try
{
    var results = Enumerable.Range(0, 100)
        .AsParallel()
        .Select(x => 100 / x)  // 除以零
        .ToList();
}
catch (AggregateException ae)
{
    foreach (var ex in ae.InnerExceptions)
    {
        Console.WriteLine($"异常: {ex.Message}");
    }
}

using System.Threading;

CancellationTokenSource cts = new CancellationTokenSource();
cts.CancelAfter(1000);

try
{
    Parallel.For(0, 10000000, new ParallelOptions 
    { 
        CancellationToken = cts.Token 
    }, i => {
        DoWork(i);
    });
}
catch (OperationCanceledException)
{
    Console.WriteLine("并行操作被取消");
}

// 不适用并行：循环体太简单
for (int i = 0; i < 1000000; i++)
{
    sum += i;  // 并行化会增加锁开销
}

// 适用并行：循环体计算密集
Parallel.For(0, 1000, i => {
    var result = Enumerable.Range(0, 1000000)
        .Select(x => Math.Sqrt(x) * Math.PI)
        .Sum();
});

// 使用 ThreadLocal 避免锁
Parallel.For(0, 1000000, () => 0, (i, state, local) => local + i, 
    local => Interlocked.Add(ref sum, local));

// 查看是否真的并行执行
var query = Enumerable.Range(0, 100)
    .AsParallel()
    .WithDegreeOfParallelism(4)
    .Select(x => {
        Console.WriteLine($"处理 {x}，线程: {Thread.CurrentThread.ManagedThreadId}");
        return x * 2;
    })
    .ToList();