按需创建线程并固定任务组至同一线程的实现问题

我需要实现按需创建线程（发现新任务时创建），并且让特定任务始终在同一线程中执行。举个例子：现有7个任务，单线程分配上限是3个，那么thread1负责执行item1、item2、item3；thread2负责item4、item5、item6；thread3负责item7，后续新增的2个任务也必须添加到thread3中。

我查过ThreadLocal和ThreadStatic属性的文档，但不知道怎么把它们用到这个场景里。找到一篇需求完全匹配的帖子，但方案不管用。下面是我目前的代码：

ConcurrentDictionary<string, List<ItemInfo>> _itemToThreadMap = new ConcurrentDictionary<string, List<ItemInfo>>(); 

const int THREAD_PARTITION_SIZE = 10;

[ThreadStatic] List<ItemInfo> itemInfoList = new List<ItemInfo>();

internal void CreateOrUpdateThreadMap()
{
    try
    { 
        ItemInfo info = new ItemInfo(); // new item
        itemInfoList.Add(info);
        string threadId = _itemToThreadMap.FirstOrDefault(t => t.Value.Count < THREAD_PARTITION_SIZE).Key;
        if (!string.IsNullOrEmpty(threadId))
        { 
            _itemToThreadMap.TryGetValue(threadId, out List<ItemInfo> itemList); 
            _itemToThreadMap.TryUpdate(threadId, itemInfoList, itemList); 
        }
        else
        { 
            string newThreadId = $"thread_{_itemToThreadMap.Count + 1}";  

            _printerToThreadMap.TryAdd(newThreadId, itemInfoList);
            SpawnNewThread(newThreadId);
        }
    }
    catch (Exception ex)
    {
        _logger.Error("Failed to create/update thread: ", ex);
    }
} 

internal void SpawnNewThread(string newThreadId)
{ 
    try
    {
        lock (lockObj)
        {
            Thread itemThread = new Thread(new ThreadStart(MonitorJobsForThread));
            printerThread.Name = newThreadId;
            printerThread.Start();
        }
    }
    catch (Exception ex)
    { 
    }
}

internal void MonitorJobsForThread()
{
   // some long running job for this thread.
}

现有代码的核心问题 #

1. [ThreadStatic]使用错误：这个属性标记的变量是每个线程独立拥有的，调用CreateOrUpdateThreadMap的线程（比如主线程）里的itemInfoList，和新创建的工作线程里的itemInfoList完全是两个不同的实例，导致线程和任务的映射完全混乱。
2. 线程映射逻辑错误：TryUpdate用当前线程的列表替换目标线程的列表，完全不符合“把新任务加到目标线程队列”的需求；另外FirstOrDefault遍历ConcurrentDictionary不是线程安全操作，可能导致数据不一致。
3. 变量名错误：_printerToThreadMap未定义，应该是_itemToThreadMap；SpawnNewThread里的printerThread应该是itemThread。
4. 任务队列非线程安全：用List<ItemInfo>做任务容器，多线程添加/读取会引发并发问题。

修正后的实现方案 #

核心思路是：维护一个线程ID与线程安全任务队列的映射，新任务优先加入未满的队列；队列满时创建新线程并绑定新队列；每个线程持续监听自己的队列，取出任务执行。

using System.Collections.Concurrent;
using System.Threading;

public class TaskThreadManager
{
    // 线程ID -> 对应任务队列（线程安全）
    private readonly ConcurrentDictionary<string, ConcurrentQueue<ItemInfo>> _threadTaskQueues = new();
    private readonly object _threadCreationLock = new();
    private const int THREAD_PARTITION_SIZE = 10;
    private bool _isRunning = true; // 线程停止信号

    // 新增任务的入口方法
    internal void AddNewTask(ItemInfo newItem)
    {
        try
        {
            // 尝试找到第一个未满的任务队列
            var targetThreadId = _threadTaskQueues
                .FirstOrDefault(kv => kv.Value.Count < THREAD_PARTITION_SIZE)
                .Key;

            if (!string.IsNullOrEmpty(targetThreadId))
            {
                // 将任务加入目标线程的队列
                _threadTaskQueues[targetThreadId].Enqueue(newItem);
                return;
            }

            // 没有未满队列，创建新线程
            lock (_threadCreationLock)
            {
                // 双重检查，避免多线程重复创建线程
                targetThreadId = _threadTaskQueues
                    .FirstOrDefault(kv => kv.Value.Count < THREAD_PARTITION_SIZE)
                    .Key;

                if (!string.IsNullOrEmpty(targetThreadId))
                {
                    _threadTaskQueues[targetThreadId].Enqueue(newItem);
                    return;
                }

                // 生成新线程ID并创建队列
                var newThreadId = $"thread_{_threadTaskQueues.Count + 1}";
                var newTaskQueue = new ConcurrentQueue<ItemInfo>();
                newTaskQueue.Enqueue(newItem);
                _threadTaskQueues.TryAdd(newThreadId, newTaskQueue);

                // 启动新线程
                SpawnNewThread(newThreadId);
            }
        }
        catch (Exception ex)
        {
            _logger.Error("Failed to add new task: ", ex);
        }
    }

    private void SpawnNewThread(string threadId)
    {
        var workerThread = new Thread(() => MonitorJobsForThread(threadId))
        {
            Name = threadId,
            IsBackground = true // 设置为后台线程，不阻止程序退出
        };
        workerThread.Start();
    }

    // 线程的任务监听逻辑
    private void MonitorJobsForThread(string threadId)
    {
        while (_isRunning)
        {
            if (_threadTaskQueues.TryGetValue(threadId, out var taskQueue) 
                && taskQueue.TryDequeue(out var itemToProcess))
            {
                // 执行任务的实际逻辑，替换成你的业务代码
                ProcessTaskItem(itemToProcess);
            }
            else
            {
                // 队列空时短暂休眠，避免空转占用CPU
                Thread.Sleep(100);
            }
        }
    }

    // 示例任务处理方法
    private void ProcessTaskItem(ItemInfo item)
    {
        // 这里写你的任务执行逻辑
        _logger.Info($"Thread {Thread.CurrentThread.Name} processing item: {item.Id}");
    }

    // 优雅停止所有线程的方法
    internal void StopAllWorkerThreads()
    {
        _isRunning = false;
    }
}

// 任务信息类（根据你的实际需求修改）
public class ItemInfo
{
    public string Id { get; set; } = Guid.NewGuid().ToString();
    // 其他业务属性...
}

关键改动说明 #

1. 线程安全队列：用ConcurrentQueue<ItemInfo>替代List<ItemInfo>，保证多线程环境下的入队/出队操作安全，无需额外加锁。
2. 移除[ThreadStatic]：任务队列直接与线程ID绑定，不存在跨线程实例混淆的问题，逻辑更清晰。
3. 双重检查锁：创建新线程时加锁并二次检查，避免多线程同时触发线程创建，导致重复线程。
4. 线程持续监听：每个线程启动后循环读取自己的任务队列，空队列时短暂休眠降低CPU占用。
5. 优雅停止：通过_isRunning信号控制线程退出，避免强制终止线程引发的问题。

内容的提问来源于stack exchange，提问作者ashwathmabiyan