企业官网建设流程全解析

视觉检测预处理流水线：从 25fps 到 100fps 的优化实录

一、背景：一个差四倍的性能指标

之前调一个视觉检测项目，相机采集帧率60fps，算法要求预处理延迟在10ms 以内。我写了一条流水线：

灰度化 → 高斯滤波 → 边缘检测 → 缩放

四个步骤依次执行，单帧耗时 40ms。差了四倍。

用 Profiler 打了一下，发现不是算法本身慢。每个步骤都在new Mat，前一秒刚分配的内存，下一秒就被 GC 回收，再下一秒又分配。GC 频繁触发，采集线程被挂起，帧率直接掉到30fps。

后来把预处理拆成了一条流水线，用Channel连接每个阶段，内存池复用缓冲区。单帧延迟从40ms 降到了 8ms。

二、问题出在"一帧做完再做下一帧"

最初的写法，每一帧从相机出来，串行跑完四个步骤，跑完了再处理下一帧。这种模式叫“流水线阻塞”——后面的帧必须等前面的帧全部处理完才能开始。

更要命的是，每一步都在 new Mat。Cv2.GaussianBlur内部会分配新的 Mat，Cv2.Resize也会。这些 Mat 是非托管内存，虽然 GC 管得松，但分配和释放本身有开销。60fps 的时候，每秒要分配几百个 Mat。

用 Visual Studio 的诊断工具看了一眼：

• 内存分配速率：每秒几百 MB
• Gen2 GC：每分钟触发好几次
• 每次触发暂停：几十毫秒

采集线程就在这几十毫秒里被挂起，丢帧。

三、流水线的思路

把预处理拆成几个独立的阶段，每个阶段跑在自己的线程上，阶段之间用队列连接。帧从相机出来，丢进第一个队列，阶段一处理完丢进第二个队列，阶段二接着处理，以此类推。

每个阶段都在独立处理自己的帧，不用等别的阶段。理想情况下，四个阶段同时在工作，吞吐量则是串行模式的4 倍。

内存池的作用：每个阶段需要一块内存来存输出，从池里借，用完还回去。不每次都 new。池子里的缓冲区在启动时一次性分配好，运行时反复复用。

四、完整实现

用System.Threading.Channels做队列，用ArrayPool管理内存。

4.1 帧数据结构

先定义帧数据结构，用一个简单的对象池管理缓冲区。

usingSystem;usingSystem.Buffers;usingSystem.Threading.Channels;usingSystem.Threading.Tasks;usingOpenCvSharp;/// <summary>/// 预处理帧：携带原始数据和中间结果/// </summary>publicclassPipelineFrame:IDisposable{publicbyte[]RawData{get;set;}// 原始图像数据publicbyte[]Stage1Data{get;set;}// 阶段1输出（灰度）publicbyte[]Stage2Data{get;set;}// 阶段2输出（滤波）publicbyte[]Stage3Data{get;set;}// 阶段3输出（边缘）publicbyte[]FinalData{get;set;}// 最终输出（缩放）publicintWidth{get;set;}publicintHeight{get;set;}publiculongTimestamp{get;set;}// 记录各阶段耗时，便于分析瓶颈publiclongCaptureToStage1Us{get;set;}publiclongStage1ToStage2Us{get;set;}publiclongStage2ToStage3Us{get;set;}publiclongStage3ToFinalUs{get;set;}publiclongTotalLatencyUs{get;set;}publicvoidDispose(){// 归还所有借用的缓冲区if(RawData!=null){ArrayPool<byte>.Shared.Return(RawData);RawData=null;}if(Stage1Data!=null){ArrayPool<byte>.Shared.Return(Stage1Data);Stage1Data=null;}if(Stage2Data!=null){ArrayPool<byte>.Shared.Return(Stage2Data);Stage2Data=null;}if(Stage3Data!=null){ArrayPool<byte>.Shared.Return(Stage3Data);Stage3Data=null;}if(FinalData!=null){ArrayPool<byte>.Shared.Return(FinalData);FinalData=null;}}}

4.2 流水线核心

接下来是流水线的核心，把四个阶段串起来。

publicclassImagePreprocessPipeline:IDisposable{// 四个阶段的队列privatereadonlyChannel<PipelineFrame>_stage1Queue;privatereadonlyChannel<PipelineFrame>_stage2Queue;privatereadonlyChannel<PipelineFrame>_stage3Queue;privatereadonlyChannel<PipelineFrame>_outputQueue;privatereadonlyint_width;privatereadonlyint_height;// 预分配的 Mat，每个阶段复用privatereadonlyMat_grayMat;privatereadonlyMat_blurMat;privatereadonlyMat_edgeMat;privatereadonlyMat_resizeMat;privatereadonlyCancellationTokenSource_cts;privatereadonlyTask[]_stageTasks;// 统计privatelong_inputCount;privatelong_outputCount;privatelong_dropCount;publicImagePreprocessPipeline(intwidth,intheight,intqueueCapacity=5){_width=width;_height=height;// 有界队列，队列满时丢弃新帧，避免无限堆积_stage1Queue=Channel.CreateBounded<PipelineFrame>(queueCapacity);_stage2Queue=Channel.CreateBounded<PipelineFrame>(queueCapacity);_stage3Queue=Channel.CreateBounded<PipelineFrame>(queueCapacity);_outputQueue=Channel.CreateBounded<PipelineFrame>(queueCapacity);// 每个阶段预分配一个 Mat_grayMat=newMat(height,width,MatType.CV_8UC1);_blurMat=newMat();_edgeMat=newMat();_resizeMat=newMat();_cts=newCancellationTokenSource();// 启动四个阶段线程_stageTasks=newTask[]{Task.Run(()=>Stage1Loop(_cts.Token)),Task.Run(()=>Stage2Loop(_cts.Token)),Task.Run(()=>Stage3Loop(_cts.Token)),Task.Run(()=>FinalLoop(_cts.Token))};}/// <summary>/// 采集端调用：入队一帧原始数据/// </summary>publicboolEnqueue(IntPtrpData,intdataSize,ulongtimestamp){// 从池里借一块内存byte[]buffer=ArrayPool<byte>.Shared.Rent(dataSize);// 拷贝相机数据到池缓冲区System.Runtime.InteropServices.Marshal.Copy(pData,buffer,0,dataSize);varframe=newPipelineFrame{RawData=buffer,Width=_width,Height=_height,Timestamp=timestamp};// 入队，队列满则丢帧if(_stage1Queue.Writer.TryWrite(frame)){System.Threading.Interlocked.Increment(ref_inputCount);returntrue;}// 队列满，归还内存并丢弃frame.Dispose();System.Threading.Interlocked.Increment(ref_dropCount);returnfalse;}

4.3 阶段一：灰度化

privateasyncTaskStage1Loop(CancellationTokenct){varreader=_stage1Queue.Reader;while(!ct.IsCancellationRequested){try{varframe=awaitreader.ReadAsync(ct);longstart=System.Diagnostics.Stopwatch.GetTimestamp();// 用 Mat 包装原始数据，零拷贝unsafe{using(varsrc=newMat(_height,_width,MatType.CV_8UC3,(IntPtr)System.Runtime.CompilerServices.Unsafe.AsPointer(refframe.RawData[0]))){// 灰度化，输出到预分配的 MatCv2.CvtColor(src,_grayMat,ColorConversionCodes.BGR2GRAY);}}// 拷贝结果到池缓冲区intgraySize=_width*_height;frame.Stage1Data=ArrayPool<byte>.Shared.Rent(graySize);System.Runtime.InteropServices.Marshal.Copy(_grayMat.Data,frame.Stage1Data,0,graySize);frame.CaptureToStage1Us=ElapsedUs(start);await_stage2Queue.Writer.WriteAsync(frame,ct);}catch(OperationCanceledException){break;}catch(Exceptionex){System.Diagnostics.Debug.WriteLine($"阶段1异常:{ex.Message}");}}}

4.4 阶段二：高斯滤波

privateasyncTaskStage2Loop(CancellationTokenct){varreader=_stage2Queue.Reader;while(!ct.IsCancellationRequested){try{varframe=awaitreader.ReadAsync(ct);longstart=System.Diagnostics.Stopwatch.GetTimestamp();// 从池缓冲区构造 Matunsafe{using(varsrc=newMat(_height,_width,MatType.CV_8UC1,(IntPtr)System.Runtime.CompilerServices.Unsafe.AsPointer(refframe.Stage1Data[0]))){// 滤波到预分配 MatCv2.GaussianBlur(src,_blurMat,newSize(5,5),1.5);}}intsize=_width*_height;frame.Stage2Data=ArrayPool<byte>.Shared.Rent(size);System.Runtime.InteropServices.Marshal.Copy(_blurMat.Data,frame.Stage2Data,0,size);frame.Stage1ToStage2Us=ElapsedUs(start);await_stage3Queue.Writer.WriteAsync(frame,ct);}catch(OperationCanceledException){break;}catch(Exceptionex){System.Diagnostics.Debug.WriteLine($"阶段2异常:{ex.Message}");}}}

4.5 阶段三：边缘检测

privateasyncTaskStage3Loop(CancellationTokenct){varreader=_stage3Queue.Reader;while(!ct.IsCancellationRequested){try{varframe=awaitreader.ReadAsync(ct);longstart=System.Diagnostics.Stopwatch.GetTimestamp();unsafe{using(varsrc=newMat(_height,_width,MatType.CV_8UC1,(IntPtr)System.Runtime.CompilerServices.Unsafe.AsPointer(refframe.Stage2Data[0]))){Cv2.Canny(src,_edgeMat,50,150);}}intsize=_width*_height;frame.Stage3Data=ArrayPool<byte>.Shared.Rent(size);System.Runtime.InteropServices.Marshal.Copy(_edgeMat.Data,frame.Stage3Data,0,size);frame.Stage2ToStage3Us=ElapsedUs(start);await_outputQueue.Writer.WriteAsync(frame,ct);}catch(OperationCanceledException){break;}catch(Exceptionex){System.Diagnostics.Debug.WriteLine($"阶段3异常:{ex.Message}");}}}

4.6 阶段四：缩放

privateasyncTaskFinalLoop(CancellationTokenct){varreader=_outputQueue.Reader;while(!ct.IsCancellationRequested){try{varframe=awaitreader.ReadAsync(ct);longstart=System.Diagnostics.Stopwatch.GetTimestamp();unsafe{using(varsrc=newMat(_height,_width,MatType.CV_8UC1,(IntPtr)System.Runtime.CompilerServices.Unsafe.AsPointer(refframe.Stage3Data[0]))){Cv2.Resize(src,_resizeMat,newSize(640,480));}}intsize=640*480;frame.FinalData=ArrayPool<byte>.Shared.Rent(size);System.Runtime.InteropServices.Marshal.Copy(_resizeMat.Data,frame.FinalData,0,size);frame.Stage3ToFinalUs=ElapsedUs(start);frame.TotalLatencyUs=ElapsedUs(0);System.Threading.Interlocked.Increment(ref_outputCount);// 输出帧交给下游处理（推理、存盘等）OnFrameProcessed?.Invoke(frame);// 处理完归还内存frame.Dispose();}catch(OperationCanceledException){break;}catch(Exceptionex){System.Diagnostics.Debug.WriteLine($"阶段4异常:{ex.Message}");}}}publicAction<PipelineFrame>OnFrameProcessed{get;set;}privatestaticlongElapsedUs(longstartTicks){longelapsedTicks=System.Diagnostics.Stopwatch.GetTimestamp()-startTicks;returnelapsedTicks*1_000_000/System.Diagnostics.Stopwatch.Frequency;}publicvoidDispose(){_cts.Cancel();_stage1Queue.Writer.TryComplete();_stage2Queue.Writer.TryComplete();_stage3Queue.Writer.TryComplete();_outputQueue.Writer.TryComplete();try{Task.WaitAll(_stageTasks,3000);}catch{}_grayMat?.Dispose();_blurMat?.Dispose();_edgeMat?.Dispose();_resizeMat?.Dispose();_cts.Dispose();}}

五、接入相机

相机回调里只做一件事——把数据拷进池缓冲区，然后入队。

publicclassCameraCapture{privateImagePreprocessPipeline_pipeline;publicvoidInitialize(){_pipeline=newImagePreprocessPipeline(2448,2048,queueCapacity:5);// 输出帧的回调，交给下游做推理_pipeline.OnFrameProcessed=(frame)=>{// 用 FinalData 做推理RunInference(frame.FinalData);};// 注册相机回调RegisterImageCallback(OnFrame);}privatevoidOnFrame(IntPtrpData,refMV_FRAME_OUT_INFO_EXinfo,IntPtruser){intdataSize=(int)(info.nWidth*info.nHeight*3);// 入队（内部会从池里借内存并拷贝）boolenqueued=_pipeline.Enqueue(pData,dataSize,info.nTimestamp);if(!enqueued){// 丢帧计数_dropCount++;}}}

六、流水线的收益和代价

收益：四个阶段并行工作，吞吐量是串行的 4 倍。内存池复用，每帧的分配从 4 次降到 1 次（只有入队时从池里借一次，后续阶段复用同一帧的多个缓冲区）。

代价：流水线有"填充"和"排空"时间。启动时四个阶段都空着，第一帧要跑完四个阶段才出结果。运行时，四个阶段各处理一帧，总延迟是四阶段耗时之和，但吞吐量是单阶段的速度。

实测数据，2448×2048 的图，四个阶段：

单帧延迟反而略增（多了队列传递的开销），但吞吐量翻了 4 倍。产线要的是吞吐量，这个账划算。

七、几个容易踩的坑

1. 队列容量不能设太大。设成 50，每帧的数据在队列里堆着，延迟会累积。设成 5 就够了，队列满了直接丢帧，保证最新的帧优先处理。

2. 每个阶段的 Mat 必须复用。Cv2.GaussianBlur如果不传输出 Mat，内部会新建一个。传预分配的 Mat 作为输出，避免反复分配。

3. ArrayPool 借的内存要还。PipelineFrame.Dispose里统一归还所有缓冲区。漏一个就内存泄漏。

4. Span 和指针不能跨线程。每个阶段在自己的线程里处理完，把数据拷进池缓冲区再入队。不要把IntPtr传到下一个阶段。

5. 入队失败要处理。队列满时TryWrite返回 false，这时候要归还内存、丢弃帧。不处理的话内存泄漏。

八、什么时候该上流水线

九、总结

本质上图像预处理流水线的核心就三件事：

1. 拆分成多个阶段，每个阶段独立线程，用 Channel 连接
2. 内存池复用，每帧只从池里借一次，各阶段复用同一帧的缓冲区
3. 有界队列，满了就丢帧，保证延迟不累积

那个视觉检测项目，改成流水线逻辑之后，吞吐量从25fps 提到了 100fps，GC 几乎不触发了。单帧延迟则从40ms 变成了 45ms，但产线要的是吞吐量，这点延迟增加完全可以接受。

小提示：第一次上流水线，先用Stopwatch记录每个阶段的耗时。如果某个阶段明显慢，说明它是瓶颈，可以给它单独加线程或者优化算法。用PipelineFrame里的各阶段时间字段，跑个几百帧导出 CSV，哪个阶段拖后腿一目了然。