Python中多线程任务队列中的常见错误与解决方案

# national_match_task.py
def get_failed_records():
    from app import app  # 延迟导入
    with app.app_context():
        records = db.session.query(CustomerOrder).filter(...).all()
    return records

# national_match_task.py
def start_processing(app):  # 接收app参数
    # 使用app而不是直接导入

# app.py
from task.national_match_task import start_processing
start_processing(app)  # 传入app实例

from flask import current_app as app  # 替代直接导入

# match_phone_number.py
def match_nationwide_numbers(item, cookie, logger):
    if not (item.prefix and item.suffix):  # 使用 . 访问属性
        logger.warning("缺少必要的前缀或后缀信息")
        return {"匹配状态": "失败: 缺少前缀或后缀"}
    # 其他逻辑...

# national_match_task.py
def worker():
    item = queue.get()
    item_dict = {
        'prefix': item.prefix,
        'suffix': item.suffix,
        'tracking_number': item.tracking_number,
    }
    result = match_nationwide_numbers(item_dict, item.cookie, logger)

max_retries = 3
retry_count = getattr(item, '_retry_count', 0)
if retry_count < max_retries:
    item._retry_count = retry_count + 1
    queue.put(item)  # 重新放回队列

def worker():
    item = None  # 初始化
    try:
        item = queue.get(timeout=1)
        # 处理逻辑...
    except queue.Empty:
        continue
    finally:
        if item is not None:  # 确保变量已赋值
            queue.task_done()

finally:
    if 'item' in locals() and item is not None:
        queue.task_done()

def worker():
    while True:
        process_next_item()

def process_next_item():
    item = queue.get(timeout=1)
    try:
        # 处理逻辑...
    finally:
        queue.task_done()

import threading
import queue
import time
from flask import current_app as app
from models import CustomerOrder

def worker():
    item = None  # 初始化
    try:
        item = queue.get(timeout=1)
        if item is None:
            return

        logger.info(f"处理记录: {item.tracking_number}")
        result = match_nationwide_numbers({
            'prefix': item.prefix,
            'suffix': item.suffix,
        }, item.cookie, logger)

        update_record(item.id, result["匹配状态"], result.get("手机号"))
        
    except queue.Empty:
        return
    except Exception as e:
        logger.error(f"处理失败: {e}")
        if item and getattr(item, '_retry_count', 0) < 3:
            item._retry_count += 1
            queue.put(item)
    finally:
        if item is not None:
            queue.task_done()

def start_processing(app):
    for _ in range(5):
        threading.Thread(target=worker, daemon=True).start()