本文深入剖析在 streamlit 多线程视频采集场景下，因 ocr 任务阻塞主线程/工作线程导致的区域检测丢失问题，并提供基于线程安全、资源隔离与异步调度的完整解决方案。
本文深入剖析在 streamlit 多线程视频采集场景下，因 ocr 任务阻塞主线程/工作线程导致的区域检测丢失问题，并提供基于线程安全、资源隔离与异步调度的完整解决方案。
在基于 Streamlit 构建的实时 OCR 应用中，开发者常采用 daemon=True 的后台线程持续捕获摄像头帧（如 video_capture_thread），并在帧就绪后调用 check_detection_region() 启动多线程 OCR 处理。然而，正如日志所示——尽管每轮均提交 10 个区域任务（Rgions: 10-10），但部分区域（如 Region-4、Region-8 等）长期未出现在 OCR 输出中，最终表现为“某些区域始终无法被检测”。根本原因并非模型精度或 ROI 裁剪错误，而是严重的线程资源泄漏与执行上下文冲突。
? 问题本质：OCR 实例跨线程复用 + 阻塞式调用引发的线程饥饿
PaddleOCR 的 ocr.ocr() 方法不是线程安全的，其内部依赖全局状态（如 ONNX Runtime session、CUDA context、模型图缓存等）。当多个 ThreadPoolExecutor 工作线程并发调用同一 ocr 实例时：
- 多个线程争抢共享资源（如推理引擎锁、内存池），导致部分线程长时间等待甚至死锁；
-
ocr.ocr()是同步阻塞调用，单次耗时达 0.2–0.8 秒（见日志Region（...）：X Text: Y, Conf: Z前的时间戳差），而视频采集线程以 ~100 FPS（time.sleep(0.01)）高频触发check_detection_region()—— 每秒新建一个ThreadPoolExecutor(max_workers=2)，却未显式关闭。这导致：-
线程泄漏：
ThreadPoolExecutor在作用域结束时不会自动shutdown(wait=True)，其内部线程可能持续挂起，占用 OS 线程资源； - 任务积压与丢弃：新一帧到来时，前一轮尚未完成的 OCR 任务被丢弃（无取消机制），造成区域检测“跳帧”；
-
状态污染：
last_process_time和last_check_time字典被多线程无锁读写（注释掉的with process_time_lock:未生效），引发时间戳错乱，进一步抑制区域重检。
-
线程泄漏：
✅ 正确实践：分离 I/O 与计算、复用 OCR 实例、显式生命周期管理
1. 全局单例 OCR 实例（线程安全初始化）
# ✅ 在模块顶层初始化一次，避免重复加载模型
import threading
from paddleocr import PaddleOCR
# 使用双重检查锁定确保线程安全初始化
_ocr_instance = None
_ocr_lock = threading.Lock()
def get_ocr_instance():
global _ocr_instance
if _ocr_instance is None:
with _ocr_lock:
if _ocr_instance is None:
_ocr_instance = PaddleOCR(
det_model_dir="./ch_PP-OCRv4_det_infer",
rec_model_dir="./ch_PP-OCRv4_rec_infer",
use_angle_cls=True,
lang="en",
log_level='ERROR',
use_gpu=True, # 显式启用 GPU 加速（若可用）
gpu_mem=2000 # 限制显存占用，防 OOM
)
return _ocr_instance2. 改造 check_detection_region()：使用固定线程池 + 任务队列
import concurrent.futures
import queue
import time
# ✅ 全局复用线程池（非每次创建），并设置优雅关闭
ocr_executor = concurrent.futures.ThreadPoolExecutor(
max_workers=2,
thread_name_prefix="paddleocr-worker"
)
# ✅ 使用线程安全队列暂存待处理帧（替代原 frame_buffer + set 机制）
frame_queue = queue.Queue(maxsize=2) # 最多缓存 1 帧，避免堆积
def video_capture_thread():
cap = cv2.VideoCapture(0)
try:
while True:
ret, frame = cap.read()
if not ret:
continue
frame = cv2.resize(frame, target_resolution)
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
# ⚠️ 非阻塞放入队列，旧帧自动丢弃
try:
frame_queue.put_nowait((frame_rgb, time.time()))
except queue.Full:
try:
frame_queue.get_nowait() # 弹出旧帧
frame_queue.put_nowait((frame_rgb, time.time()))
except:
pass
time.sleep(0.03) # 降低采集频率至 ~30 FPS，匹配 OCR 能力
finally:
cap.release()
# ✅ 主检测循环：从队列取帧，提交 OCR 任务（不阻塞采集线程）
def detection_loop():
last_frame_time = 0
while True:
try:
frame_rgb, capture_time = frame_queue.get(timeout=1.0)
# 防止过期帧（>500ms）参与 OCR
if time.time() - capture_time > 0.5:
continue
# ✅ 提交异步 OCR 任务，立即返回
ocr_executor.submit(
process_frame_batch,
frame_rgb,
current_detection_region,
last_process_time,
last_check_time,
get_ocr_instance()
)
except queue.Empty:
continue3. 关键修复：process_region 中移除竞态操作，添加超时与错误隔离
def process_region(region_idx, region, frame, region_data, last_process_time,
last_check_time, ocr_inst):
# ✅ 使用 region_idx 作为唯一键，避免字典竞态
rd = region_data.setdefault(region_idx, {
'detection_cache': [],
'last_cache_clear': time.time(),
'datetime_info': '',
'last_process_time': 0.0
})
x1, y1, x2, y2 = preprocess_bbox(region, frame)
roi = frame[y1:y2, x1:x2]
if roi.size == 0:
return None
try:
# ✅ 设置 OCR 超时（防卡死）
ocr_result = ocr_inst.ocr(roi, cls=True, det=False) # 可选：禁用检测，仅识别
if not ocr_result:
return None
# ✅ 单区域结果处理（轻量，避免长耗时）
best_match = None
for line in ocr_result:
for detection in line:
text, conf = detection[1]
if conf > 0.7 and re.match(r"^\d+(\.\d+)?$", text.strip()):
num_val = float(change_num(region=region_idx, text=text.strip()))
if best_match is None or conf > best_match['conf']:
best_match = {
'match_text': num_val,
'conf': conf,
'raw_text': text.strip(),
'time': time.time()
}
if best_match:
rd['detection_cache'].append(best_match)
# ✅ 清理缓存逻辑移到主函数，此处只追加
return {
'region_id': region_idx,
'value': best_match,
'timestamp': time.time()
}
except Exception as e:
print(f"[OCR Error R{region_idx}] {e}")
return None
# ✅ 批处理入口：统一管理缓存与时间戳
def process_frame_batch(frame, regions, last_proc, last_check, ocr_inst):
results = []
current_time = time.time()
for idx, region in enumerate(regions):
# ✅ 原子化更新时间戳（使用 dict.setdefault 保证线程安全）
last_proc.setdefault(idx, 0.0)
last_check.setdefault(idx, 0.0)
if current_time - last_proc[idx] < 1.0: # 全局最小间隔 1s
continue
last_proc[idx] = current_time
result = process_region(idx, region, frame, {}, last_proc, last_check, ocr_inst)
if result:
results.append(result)
if results:
# ✅ 统一后处理：去重、取最优、触发 UI 更新（通过 Streamlit session state）
final_result = aggregate_results(results)
st.session_state.ocr_results = final_result # 示例：推送至 Streamlit 状态? 关键注意事项总结
-
绝不复用
ThreadPoolExecutor实例：每次with ThreadPoolExecutor()都会新建线程池，务必改为全局单例 +shutdown()配合应用生命周期； - OCR 实例必须全局单例：PaddleOCR 初始化开销大且非线程安全，禁止在多线程中重复构造；
-
帧队列需限流：使用
queue.Queue(maxsize=N)防止内存溢出，配合put_nowait()/get_nowait()实现“最新帧优先”策略； -
显式超时控制：为
ocr.ocr()添加timeout参数（若版本支持）或用concurrent.futures.wait()限制总耗时； -
Streamlit 状态同步：OCR 结果应通过
st.session_state安全传递至前端，避免直接操作全局变量引发渲染竞争。
通过以上重构，OCR 任务将脱离视频采集线程的强耦合，实现稳定、可预测的区域检测能力——日志中缺失的 Region 将不再“神秘消失”，系统吞吐量与可靠性显著提升。
