线程亲和性

知识点

什么是线程亲和性

线程亲和性（Thread Affinity）是指将线程绑定到特定的CPU核心或处理器上运行的技术。通过设置线程亲和性，可以控制线程在哪个CPU核心上执行，从而优化性能、减少缓存失效和提高系统的可预测性。

线程亲和性的作用

1. 缓存优化：减少缓存失效，提高缓存命中率
2. NUMA优化：在NUMA架构中优化内存访问
3. 实时性：提高实时应用的可预测性
4. 热点避免：避免某些核心过度使用
5. 调试优化：便于性能分析和调试

ProcessorAffinity属性

在.NET中，可以通过Thread.ProcessorAffinity属性（在.NET Framework中）或使用P/Invoke调用Windows API来设置线程亲和性。

应用场景

• 高性能计算应用
• 实时系统
• 游戏引擎
• 音视频处理
• 网络服务器优化

注意事项

• 不当使用可能降低性能
• 操作系统调度器通常做得很好
• 应该基于性能测试做决定
• 考虑NUMA架构的影响

代码案例

案例1：基本线程亲和性设置

using System;
using System.Diagnostics;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;

class BasicThreadAffinity
{
    // Windows API声明
    [DllImport("kernel32.dll")]
    static extern IntPtr GetCurrentThread();
    
    [DllImport("kernel32.dll")]
    static extern IntPtr SetThreadAffinityMask(IntPtr hThread, IntPtr dwThreadAffinityMask);
    
    [DllImport("kernel32.dll")]
    static extern int GetCurrentThreadId();
    
    static void Main(string[] args)
    {
        Console.WriteLine("线程亲和性基本示例");
        Console.WriteLine($"系统CPU核心数: {Environment.ProcessorCount}");
        
        // 显示当前进程的亲和性
        var process = Process.GetCurrentProcess();
        Console.WriteLine($"进程亲和性掩码: 0x{process.ProcessorAffinity.ToInt64():X}");
        
        // 启动多个任务，分别绑定到不同的CPU核心
        Task[] tasks = new Task[Math.Min(4, Environment.ProcessorCount)];
        
        for (int i = 0; i < tasks.Length; i++)
        {
            int coreId = i;
            tasks[i] = Task.Run(() => WorkerWithAffinity(coreId));
        }
        
        // 启动一个不设置亲和性的对比任务
        Task normalTask = Task.Run(() => NormalWorker());
        
        Task.WaitAll(tasks.Concat(new[] { normalTask }).ToArray());
        
        Console.WriteLine("所有任务完成");
    }
    
    static void WorkerWithAffinity(int coreId)
    {
        int threadId = GetCurrentThreadId();
        Console.WriteLine($"工作线程{coreId}: 线程ID={threadId}, 尝试绑定到CPU核心{coreId}");
        
        try
        {
            // 设置线程亲和性到指定核心
            IntPtr affinityMask = new IntPtr(1 << coreId);
            IntPtr previousAffinity = SetThreadAffinityMask(GetCurrentThread(), affinityMask);
            
            if (previousAffinity != IntPtr.Zero)
            {
                Console.WriteLine($"工作线程{coreId}: 成功绑定到CPU核心{coreId}");
            }
            else
            {
                Console.WriteLine($"工作线程{coreId}: 绑定到CPU核心{coreId}失败");
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"工作线程{coreId}: 设置亲和性异常 - {ex.Message}");
        }
        
        // 执行计算密集型工作
        PerformIntensiveWork($"绑定核心{coreId}", 3000);
    }
    
    static void NormalWorker()
    {
        int threadId = GetCurrentThreadId();
        Console.WriteLine($"普通工作线程: 线程ID={threadId}, 使用系统默认调度");
        
        // 执行相同的计算密集型工作
        PerformIntensiveWork("系统调度", 3000);
    }
    
    static void PerformIntensiveWork(string workerName, int durationMs)
    {
        var stopwatch = Stopwatch.StartNew();
        long operations = 0;
        
        Console.WriteLine($"{workerName}: 开始计算密集型工作");
        
        while (stopwatch.ElapsedMilliseconds < durationMs)
        {
            // 执行一些计算密集型操作
            for (int i = 0; i < 10000; i++)
            {
                Math.Sqrt(i * i + 1);
                operations++;
            }
            
            // 定期报告进度
            if (operations % 100000 == 0)
            {
                Console.WriteLine($"{workerName}: 已执行 {operations} 次操作, " +
                                $"当前线程ID: {GetCurrentThreadId()}");
            }
        }
        
        stopwatch.Stop();
        Console.WriteLine($"{workerName}: 工作完成, 总操作数: {operations}, " +
                        $"耗时: {stopwatch.ElapsedMilliseconds}ms, " +
                        $"操作/秒: {operations * 1000.0 / stopwatch.ElapsedMilliseconds:F0}");
    }
}

案例2：NUMA感知的任务分配

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;

class NumaAwareTaskDistribution
{
    [DllImport("kernel32.dll")]
    static extern bool GetNumaHighestNodeNumber(out uint HighestNodeNumber);
    
    [DllImport("kernel32.dll")]
    static extern bool GetNumaNodeProcessorMask(uint Node, out ulong ProcessorMask);
    
    [DllImport("kernel32.dll")]
    static extern IntPtr SetThreadAffinityMask(IntPtr hThread, IntPtr dwThreadAffinityMask);
    
    [DllImport("kernel32.dll")]
    static extern IntPtr GetCurrentThread();
    
    static void Main(string[] args)
    {
        Console.WriteLine("NUMA感知任务分配示例");
        
        // 获取NUMA信息
        var numaInfo = GetNumaInformation();
        if (numaInfo.Count == 0)
        {
            Console.WriteLine("系统不支持NUMA或获取NUMA信息失败");
            return;
        }
        
        Console.WriteLine($"检测到 {numaInfo.Count} 个NUMA节点:");
        foreach (var node in numaInfo)
        {
            Console.WriteLine($"  节点 {node.Key}: 处理器掩码 0x{node.Value:X}");
        }
        
        // 在每个NUMA节点上启动任务
        List<Task> tasks = new List<Task>();
        
        foreach (var numaNode in numaInfo)
        {
            int nodeId = numaNode.Key;
            ulong processorMask = numaNode.Value;
            
            // 在当前NUMA节点上启动多个任务
            int tasksPerNode = Math.Min(4, GetCoreCountFromMask(processorMask));
            
            for (int i = 0; i < tasksPerNode; i++)
            {
                int taskIndex = i;
                tasks.Add(Task.Run(() => NumaAwareWorker(nodeId, processorMask, taskIndex)));
            }
        }
        
        // 启动对比任务（不设置亲和性）
        Task normalTask = Task.Run(() => NormalTask());
        tasks.Add(normalTask);
        
        Task.WaitAll(tasks.ToArray());
        Console.WriteLine("所有任务完成");
    }
    
    static Dictionary<int, ulong> GetNumaInformation()
    {
        var numaInfo = new Dictionary<int, ulong>();
        
        try
        {
            if (GetNumaHighestNodeNumber(out uint highestNode))
            {
                for (uint nodeId = 0; nodeId <= highestNode; nodeId++)
                {
                    if (GetNumaNodeProcessorMask(nodeId, out ulong processorMask))
                    {
                        if (processorMask != 0)
                        {
                            numaInfo.Add((int)nodeId, processorMask);
                        }
                    }
                }
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"获取NUMA信息失败: {ex.Message}");
        }
        
        return numaInfo;
    }
    
    static int GetCoreCountFromMask(ulong mask)
    {
        int count = 0;
        while (mask != 0)
        {
            count += (int)(mask & 1);
            mask >>= 1;
        }
        return count;
    }
    
    static void NumaAwareWorker(int numaNodeId, ulong processorMask, int taskIndex)
    {
        Console.WriteLine($"NUMA任务 节点{numaNodeId}-{taskIndex}: 开始执行");
        
        try
        {
            // 设置线程亲和性到NUMA节点的处理器
            IntPtr affinityMask = new IntPtr((long)processorMask);
            SetThreadAffinityMask(GetCurrentThread(), affinityMask);
            Console.WriteLine($"NUMA任务 节点{numaNodeId}-{taskIndex}: 绑定到处理器掩码 0x{processorMask:X}");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"NUMA任务 节点{numaNodeId}-{taskIndex}: 设置亲和性失败 - {ex.Message}");
        }
        
        // 分配一些内存并进行计算
        PerformNumaAwareWork($"NUMA节点{numaNodeId}-任务{taskIndex}");
    }
    
    static void NormalTask()
    {
        Console.WriteLine("普通任务: 使用系统默认调度");
        PerformNumaAwareWork("普通任务");
    }
    
    static void PerformNumaAwareWork(string taskName)
    {
        var stopwatch = Stopwatch.StartNew();
        
        // 分配大量内存进行操作
        const int arraySize = 1000000;
        double[] data = new double[arraySize];
        
        Console.WriteLine($"{taskName}: 开始内存密集型计算");
        
        // 初始化数据
        for (int i = 0; i < arraySize; i++)
        {
            data[i] = Math.Sin(i * 0.001);
        }
        
        // 执行多轮计算
        for (int round = 0; round < 5; round++)
        {
            double sum = 0;
            for (int i = 0; i < arraySize; i++)
            {
                data[i] = Math.Sqrt(data[i] * data[i] + 1);
                sum += data[i];
            }
            
            Console.WriteLine($"{taskName}: 第{round + 1}轮完成, 校验和: {sum:F2}");
        }
        
        stopwatch.Stop();
        Console.WriteLine($"{taskName}: 工作完成, 耗时: {stopwatch.ElapsedMilliseconds}ms");
    }
}

案例3：高频交易系统模拟

using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;

class HighFrequencyTradingSimulation
{
    [DllImport("kernel32.dll")]
    static extern IntPtr SetThreadAffinityMask(IntPtr hThread, IntPtr dwThreadAffinityMask);
    
    [DllImport("kernel32.dll")]
    static extern IntPtr GetCurrentThread();
    
    [DllImport("kernel32.dll")]
    static extern bool SetThreadPriority(IntPtr hThread, int nPriority);
    
    const int THREAD_PRIORITY_TIME_CRITICAL = 15;
    
    private static readonly ConcurrentQueue<MarketData> marketDataQueue = new ConcurrentQueue<MarketData>();
    private static readonly ConcurrentQueue<Order> orderQueue = new ConcurrentQueue<Order>();
    private static volatile bool isRunning = true;
    
    class MarketData
    {
        public string Symbol { get; set; }
        public decimal Price { get; set; }
        public long Timestamp { get; set; }
        public int Volume { get; set; }
    }
    
    class Order
    {
        public string Symbol { get; set; }
        public decimal Price { get; set; }
        public int Quantity { get; set; }
        public long Timestamp { get; set; }
        public string Side { get; set; } // "BUY" or "SELL"
    }
    
    static void Main(string[] args)
    {
        Console.WriteLine("高频交易系统模拟 - 线程亲和性优化");
        Console.WriteLine($"系统CPU核心数: {Environment.ProcessorCount}");
        
        // 核心分配策略:
        // 核心0: 市场数据接收 (最高优先级)
        // 核心1: 策略引擎 (高优先级)
        // 核心2: 订单执行 (高优先级)
        // 核心3: 风险管理 (中等优先级)
        
        Task[] criticalTasks = new Task[4];
        
        // 市场数据接收线程 - 绑定到核心0
        criticalTasks[0] = Task.Run(() => MarketDataReceiver(0));
        
        // 策略引擎线程 - 绑定到核心1
        criticalTasks[1] = Task.Run(() => TradingStrategy(1));
        
        // 订单执行线程 - 绑定到核心2
        criticalTasks[2] = Task.Run(() => OrderExecutor(2));
        
        // 风险管理线程 - 绑定到核心3
        criticalTasks[3] = Task.Run(() => RiskManager(3));
        
        // 运行10秒
        Thread.Sleep(10000);
        isRunning = false;
        
        Task.WaitAll(criticalTasks);
        Console.WriteLine("交易系统停止");
    }
    
    static void MarketDataReceiver(int coreId)
    {
        SetupCriticalThread("市场数据接收", coreId);
        
        var random = new Random();
        string[] symbols = { "AAPL", "GOOGL", "MSFT", "TSLA", "AMZN" };
        
        while (isRunning)
        {
            // 模拟接收市场数据
            var marketData = new MarketData
            {
                Symbol = symbols[random.Next(symbols.Length)],
                Price = 100 + (decimal)(random.NextDouble() * 50),
                Timestamp = Stopwatch.GetTimestamp(),
                Volume = random.Next(100, 1000)
            };
            
            marketDataQueue.Enqueue(marketData);
            
            // 高频数据，每毫秒多个数据点
            Thread.Sleep(1);
        }
        
        Console.WriteLine("市场数据接收线程退出");
    }
    
    static void TradingStrategy(int coreId)
    {
        SetupCriticalThread("策略引擎", coreId);
        
        var random = new Random();
        decimal lastPrice = 100;
        
        while (isRunning)
        {
            if (marketDataQueue.TryDequeue(out MarketData data))
            {
                // 简单的均值回归策略
                decimal priceDiff = data.Price - lastPrice;
                
                if (Math.Abs(priceDiff) > 2) // 价格变动超过阈值
                {
                    var order = new Order
                    {
                        Symbol = data.Symbol,
                        Price = data.Price,
                        Quantity = 100,
                        Timestamp = Stopwatch.GetTimestamp(),
                        Side = priceDiff > 0 ? "SELL" : "BUY" // 均值回归
                    };
                    
                    orderQueue.Enqueue(order);
                    Console.WriteLine($"策略: 生成{order.Side}订单 {order.Symbol} @ {order.Price:F2}");
                }
                
                lastPrice = data.Price;
            }
            
            // 策略计算需要非常快
            Thread.Sleep(0); // 让出时间片但保持高响应性
        }
        
        Console.WriteLine("策略引擎线程退出");
    }
    
    static void OrderExecutor(int coreId)
    {
        SetupCriticalThread("订单执行", coreId);
        
        int executedOrders = 0;
        
        while (isRunning || !orderQueue.IsEmpty)
        {
            if (orderQueue.TryDequeue(out Order order))
            {
                // 模拟订单执行延迟
                var executionTime = Stopwatch.GetTimestamp();
                var latency = (executionTime - order.Timestamp) * 1000000.0 / Stopwatch.Frequency;
                
                Console.WriteLine($"执行: {order.Side} {order.Quantity} {order.Symbol} @ {order.Price:F2}, " +
                                $"延迟: {latency:F1}μs");
                
                executedOrders++;
                
                // 订单执行需要极低延迟
                Thread.Sleep(0);
            }
            else
            {
                Thread.Sleep(1);
            }
        }
        
        Console.WriteLine($"订单执行线程退出，总执行订单: {executedOrders}");
    }
    
    static void RiskManager(int coreId)
    {
        SetupCriticalThread("风险管理", coreId);
        
        decimal totalExposure = 0;
        const decimal maxExposure = 100000;
        
        while (isRunning)
        {
            // 模拟风险检查
            if (orderQueue.Count > 10)
            {
                Console.WriteLine("风险: 订单队列积压，可能存在风险");
            }
            
            if (totalExposure > maxExposure * 0.8m)
            {
                Console.WriteLine($"风险: 总敞口接近限制 ({totalExposure:F0}/{maxExposure:F0})");
            }
            
            // 风险管理可以稍微低频一些
            Thread.Sleep(100);
        }
        
        Console.WriteLine("风险管理线程退出");
    }
    
    static void SetupCriticalThread(string threadName, int coreId)
    {
        try
        {
            // 设置线程亲和性
            IntPtr affinityMask = new IntPtr(1 << coreId);
            SetThreadAffinityMask(GetCurrentThread(), affinityMask);
            
            // 设置高优先级
            SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_TIME_CRITICAL);
            
            Console.WriteLine($"{threadName}: 绑定到CPU核心{coreId}，设置高优先级");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"{threadName}: 设置失败 - {ex.Message}");
        }
    }
}

案例4：多媒体处理管道

using System;
using System.Diagnostics;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;

class MultimediaPipeline
{
    [DllImport("kernel32.dll")]
    static extern IntPtr SetThreadAffinityMask(IntPtr hThread, IntPtr dwThreadAffinityMask);
    
    [DllImport("kernel32.dll")]
    static extern IntPtr GetCurrentThread();
    
    private static volatile bool isProcessing = true;
    private static readonly object frameBuffer = new object();
    private static int frameCount = 0;
    
    class Frame
    {
        public int FrameNumber { get; set; }
        public byte[] Data { get; set; }
        public long Timestamp { get; set; }
    }
    
    static void Main(string[] args)
    {
        Console.WriteLine("多媒体处理管道 - 线程亲和性优化");
        Console.WriteLine("模拟实时视频处理：捕获 -> 解码 -> 处理 -> 编码 -> 输出");
        
        if (Environment.ProcessorCount < 4)
        {
            Console.WriteLine("建议至少4个CPU核心以获得最佳性能");
        }
        
        // 管道阶段分配到不同核心
        // 核心0: 视频捕获
        // 核心1: 解码
        // 核心2: 图像处理
        // 核心3: 编码和输出
        
        Task[] pipelineTasks = new Task[4];
        
        pipelineTasks[0] = Task.Run(() => VideoCaptureStage(0));
        pipelineTasks[1] = Task.Run(() => DecodeStage(1));
        pipelineTasks[2] = Task.Run(() => ImageProcessingStage(2));
        pipelineTasks[3] = Task.Run(() => EncodeOutputStage(3));
        
        // 运行8秒
        Thread.Sleep(8000);
        isProcessing = false;
        
        Task.WaitAll(pipelineTasks);
        Console.WriteLine("多媒体处理管道停止");
    }
    
    static void VideoCaptureStage(int coreId)
    {
        SetThreadAffinity("视频捕获", coreId);
        
        var stopwatch = Stopwatch.StartNew();
        int capturedFrames = 0;
        
        while (isProcessing)
        {
            // 模拟30fps视频捕获
            var targetTime = capturedFrames * 1000.0 / 30.0; // 30fps
            var currentTime = stopwatch.ElapsedMilliseconds;
            
            if (currentTime >= targetTime)
            {
                // 模拟帧捕获
                var frame = new Frame
                {
                    FrameNumber = capturedFrames,
                    Data = new byte[1920 * 1080 * 3], // 1080p RGB
                    Timestamp = Stopwatch.GetTimestamp()
                };
                
                // 填充模拟数据
                for (int i = 0; i < frame.Data.Length; i += 3)
                {
                    frame.Data[i] = (byte)(capturedFrames % 256);     // R
                    frame.Data[i + 1] = (byte)((capturedFrames * 2) % 256); // G
                    frame.Data[i + 2] = (byte)((capturedFrames * 3) % 256); // B
                }
                
                Console.WriteLine($"捕获: 帧 {capturedFrames}, 时间戳: {frame.Timestamp}");
                capturedFrames++;
                
                // 模拟捕获处理时间
                Thread.Sleep(2);
            }
            else
            {
                // 等待下一个帧时间
                Thread.Sleep(1);
            }
        }
        
        Console.WriteLine($"视频捕获完成，总捕获帧数: {capturedFrames}");
    }
    
    static void DecodeStage(int coreId)
    {
        SetThreadAffinity("解码", coreId);
        
        int decodedFrames = 0;
        
        while (isProcessing)
        {
            // 模拟解码处理
            // 在实际应用中，这里会从捕获阶段接收压缩数据并解码
            
            if (frameCount > decodedFrames)
            {
                Console.WriteLine($"解码: 处理帧 {decodedFrames}");
                
                // 模拟解码时间 (解码通常比捕获慢)
                Thread.Sleep(5);
                
                decodedFrames++;
            }
            else
            {
                Thread.Sleep(1);
            }
        }
        
        Console.WriteLine($"解码完成，总解码帧数: {decodedFrames}");
    }
    
    static void ImageProcessingStage(int coreId)
    {
        SetThreadAffinity("图像处理", coreId);
        
        int processedFrames = 0;
        
        while (isProcessing)
        {
            // 模拟图像处理 (滤镜、增强、特效等)
            if (frameCount > processedFrames)
            {
                Console.WriteLine($"处理: 应用特效到帧 {processedFrames}");
                
                // 模拟复杂的图像处理算法
                PerformImageProcessing(processedFrames);
                
                processedFrames++;
            }
            else
            {
                Thread.Sleep(1);
            }
        }
        
        Console.WriteLine($"图像处理完成，总处理帧数: {processedFrames}");
    }
    
    static void EncodeOutputStage(int coreId)
    {
        SetThreadAffinity("编码输出", coreId);
        
        int encodedFrames = 0;
        
        while (isProcessing)
        {
            // 模拟编码和输出
            if (frameCount > encodedFrames)
            {
                Console.WriteLine($"编码: 压缩并输出帧 {encodedFrames}");
                
                // 模拟编码时间 (编码通常是最耗时的)
                Thread.Sleep(8);
                
                encodedFrames++;
            }
            else
            {
                Thread.Sleep(1);
            }
        }
        
        Console.WriteLine($"编码输出完成，总输出帧数: {encodedFrames}");
    }
    
    static void SetThreadAffinity(string stageName, int coreId)
    {
        try
        {
            IntPtr affinityMask = new IntPtr(1 << coreId);
            SetThreadAffinityMask(GetCurrentThread(), affinityMask);
            
            Console.WriteLine($"{stageName}: 绑定到CPU核心{coreId}");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"{stageName}: 设置线程亲和性失败 - {ex.Message}");
        }
    }
    
    static void PerformImageProcessing(int frameNumber)
    {
        // 模拟CPU密集型图像处理
        const int iterations = 50000;
        double result = 0;
        
        for (int i = 0; i < iterations; i++)
        {
            // 模拟滤镜计算
            result += Math.Sin(i * 0.001 + frameNumber * 0.1) * Math.Cos(i * 0.002);
        }
        
        // 模拟内存操作 (像素处理)
        byte[] pixelBuffer = new byte[1024];
        for (int i = 0; i < pixelBuffer.Length; i++)
        {
            pixelBuffer[i] = (byte)((result + frameNumber + i) % 256);
        }
    }
}

知识点总结

1. 线程亲和性的核心概念：

   • 将线程绑定到特定CPU核心
   • 优化缓存命中率和内存访问
   • 提高系统可预测性

2. 适用场景：

   • 高性能计算应用
   • 实时系统和游戏
   • 多媒体处理
   • 高频交易系统

3. 设置方法：

   • Windows: SetThreadAffinityMask API
   • Linux: sched_setaffinity
   • .NET: Process.ProcessorAffinity

4. 性能考虑：

   • 合理分配任务到不同核心
   • 考虑NUMA架构影响
   • 避免核心负载不均衡

5. 最佳实践：

   • 基于性能测试做决定
   • 考虑任务特性和依赖关系
   • 留出核心给操作系统调度
   • 监控和调整策略

6. 注意事项：

   • 不当使用可能降低性能
   • 操作系统调度器通常很智能
   • 需要考虑系统整体负载
   • 跨平台兼容性问题