C#中带并发限制的并行执行：重用重型类实例实现高效图片检测

针对你100万+图片路径的异步目标检测场景，核心需求是限制并发数（如12）+ 重用高初始化成本的OnnxRuntime实例，以下是三种可行的实现方案：

方案一：使用ObjectPool（推荐） #

借助Microsoft.Extensions.ObjectPool实现实例池化，配合Parallel.ForEachAsync控制并发数，自动管理实例的创建、重用与回收。

步骤与代码 #

1. 安装NuGet依赖：Microsoft.Extensions.ObjectPool
2. 实现代码：

using Microsoft.Extensions.ObjectPool;
using Microsoft.ML.OnnxRuntime;

// 配置实例池：最大容量等于并发数（12）
var poolPolicy = new DefaultPooledPolicy<InferenceSession>
{
    Create = () => new InferenceSession("your-model-path.onnx"), // 替换为你的模型路径
    Return = session => 
    {
        // 可选：清理实例临时状态（不要Dispose，留作重用）
    }
};
var sessionPool = new DefaultObjectPool<InferenceSession>(poolPolicy, maxCapacity: 12);

// 图片路径枚举（支持IEnumerable<string>或IAsyncEnumerable<string>）
var imagePaths = GetAllImagePaths(); // 自行实现获取路径的逻辑

var parallelOpts = new ParallelOptions
{
    MaxDegreeOfParallelism = 12
};

// 并行处理所有图片路径
await Parallel.ForEachAsync(imagePaths, parallelOpts, async (path, token) =>
{
    var session = sessionPool.Get();
    try
    {
        // 执行异步目标检测（需调用OnnxRuntime的RunAsync方法）
        await RunDetectionAsync(session, path, token);
    }
    finally
    {
        // 将实例放回池，供其他任务重用
        sessionPool.Return(session);
    }
});

// 任务完成后手动清理所有实例
for (int i = 0; i < 12; i++)
{
    sessionPool.Get()?.Dispose();
}

// 异步检测逻辑示例
async Task RunDetectionAsync(InferenceSession session, string imagePath, CancellationToken token)
{
    // 加载图片并预处理
    var inputTensor = await PreprocessImageAsync(imagePath, token);
    // 异步执行推理
    using var outputs = await session.RunAsync(new List<NamedOnnxValue> { NamedOnnxValue.CreateFromTensor("input", inputTensor) }, token);
    // 处理推理结果
    ProcessDetectionResults(outputs);
}

优势 #

• 封装完善，自动处理实例的创建与重用，无需手动管理池状态
• Parallel.ForEachAsync原生支持异步枚举，适配100万+数据量的场景
• 严格控制并发数与实例数量，避免资源过载

方案二：使用分区器（Partitioner） #

将图片路径分成12个分区，每个分区绑定一个独立的InferenceSession实例，分区内串行处理、分区间并行执行，彻底避免实例在任务间切换的开销。

代码实现 #

using System.Collections.Concurrent;
using Microsoft.ML.OnnxRuntime;

var imagePaths = GetAllImagePaths();
// 创建无缓冲分区器，确保数据均匀分配
var partitioner = Partitioner.Create(imagePaths, EnumerablePartitionerOptions.NoBuffering);
var partitionTasks = new List<Task>();

// 为每个分区创建独立任务与实例
foreach (var partition in partitioner.GetPartitions(12))
{
    partitionTasks.Add(Task.Run(async () =>
    {
        // 每个分区初始化一个实例（仅初始化一次）
        using var session = new InferenceSession("your-model-path.onnx");
        try
        {
            // 串行处理分区内的所有路径
            while (partition.MoveNext())
            {
                await RunDetectionAsync(session, partition.Current, CancellationToken.None);
            }
        }
        finally
        {
            partition.Dispose();
        }
    }));
}

// 等待所有分区任务完成
await Task.WhenAll(partitionTasks);

优势 #

• 实例生命周期与分区绑定，无实例获取/放回的额外开销
• 分区内串行处理，完全符合InferenceSession不可跨并发使用的要求
• 无需额外NuGet依赖，原生API即可实现

方案三：SemaphoreSlim + ConcurrentBag（手动管理） #

如果不想引入外部依赖，可通过SemaphoreSlim控制并发数，配合ConcurrentBag手动维护实例池。

代码实现 #

using System.Collections.Concurrent;
using Microsoft.ML.OnnxRuntime;

var semaphore = new SemaphoreSlim(12, 12);
var sessionPool = new ConcurrentBag<InferenceSession>();

// 预初始化12个实例，避免运行时动态创建
for (int i = 0; i < 12; i++)
{
    sessionPool.Add(new InferenceSession("your-model-path.onnx"));
}

var imagePaths = GetAllImagePaths();
var taskList = new List<Task>();

foreach (var path in imagePaths)
{
    taskList.Add(Task.Run(async () =>
    {
        await semaphore.WaitAsync();
        InferenceSession session = null;
        try
        {
            // 从池获取实例（兜底：若池为空则临时创建）
            if (!sessionPool.TryTake(out session))
            {
                session = new InferenceSession("your-model-path.onnx");
            }
            await RunDetectionAsync(session, path, CancellationToken.None);
        }
        finally
        {
            // 将实例放回池
            if (session != null)
            {
                sessionPool.Add(session);
            }
            semaphore.Release();
        }
    }));
}

await Task.WhenAll(taskList);

// 清理所有实例
foreach (var session in sessionPool)
{
    session.Dispose();
}

优势 #

• 无外部依赖，完全使用.NET原生API
• 手动控制粒度更灵活，适合自定义扩展场景

关键注意事项 #

1. 异步操作优先：必须使用InferenceSession的RunAsync方法，避免阻塞线程导致并发效率下降
2. 实例线程安全：所有方案均保证单个实例同一时间仅被一个异步任务使用，完全符合OnnxRuntime的限制
3. 资源清理：所有InferenceSession实例最终需手动Dispose，避免内存泄漏

内容的提问来源于stack exchange，提问作者ev-dev