celery任务改为异步访问数据库
This commit is contained in:
mark.tian
@@ -16,7 +16,7 @@ from app.callback_service import (
|
||||
mark_callback_log_completed,
|
||||
get_related_records_by_unique_data_list
|
||||
)
|
||||
from app.redis_lock import redis_manager
|
||||
from app.redis_lock import redis_manager, async_redis_manager, distributed_lock
|
||||
import requests
|
||||
from app.api_config import API_CONFIG, RETRY_CONFIG
|
||||
|
||||
@@ -33,11 +33,18 @@ def conditional_task(enabled=True):
|
||||
return wrapper
|
||||
return decorator
|
||||
|
||||
# 创建同步数据库连接用于Celery任务
|
||||
engine = create_engine(
|
||||
# 创建异步数据库连接用于Celery任务
|
||||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
||||
|
||||
# 使用异步数据库连接
|
||||
async_engine = create_async_engine(
|
||||
settings.database_url,
|
||||
echo=settings.debug,
|
||||
future=True
|
||||
future=True,
|
||||
)
|
||||
|
||||
AsyncSessionLocal = async_sessionmaker(
|
||||
async_engine, class_=AsyncSession, expire_on_commit=False
|
||||
)
|
||||
|
||||
@celery_app.task(bind=True, name='push_data_to_dtc')
|
||||
@@ -47,178 +54,194 @@ def push_data_to_dtc_task(self):
|
||||
推送数据给DTC的Celery任务
|
||||
自动获取一条未完成的回调请求进行处理
|
||||
"""
|
||||
task_name = 'push_data_to_dtc'
|
||||
logger.info(f"🌿 开始推送数据给DTC任务")
|
||||
|
||||
try:
|
||||
# 获取分布式锁，使用任务名称作为锁标识
|
||||
connection_success = redis_manager.connect()
|
||||
if not connection_success:
|
||||
logger.error(f"❌ Redis管理器连接失败，任务停止执行")
|
||||
raise Exception("Redis连接失败")
|
||||
import asyncio
|
||||
from functools import partial
|
||||
|
||||
async def async_task():
|
||||
task_name = 'push_data_to_dtc'
|
||||
logger.info(f"🌿 开始推送数据给DTC任务")
|
||||
|
||||
lock = redis_manager.create_lock(f"celery_task:{task_name}", timeout=300) # 5分钟超时
|
||||
# 尝试获取锁
|
||||
if not lock.acquire(blocking=False):
|
||||
logger.warning(f"⚠️ 任务 {task_name} 正在执行中，跳过本次执行")
|
||||
return {"status": "skipped", "message": f"任务 {task_name} 正在执行中，跳过本次执行"}
|
||||
|
||||
logger.info(f"🔒 成功获取任务 {task_name} 的分布式锁")
|
||||
|
||||
with engine.connect() as conn:
|
||||
|
||||
logger.info("1")
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 0, 'total': 100, 'status': f'正在获取下一条回调请求日志...'}
|
||||
)
|
||||
logger.info("2")
|
||||
# 获取一条未完成的回调请求（按创建时间取最小值）
|
||||
query_success, callback_log_data = get_uncompleted_callback_log(conn)
|
||||
if not query_success:
|
||||
logger.error("❌ 查询未完成的回调请求失败，尝试再查一次")
|
||||
query_success, callback_log_data = get_uncompleted_callback_log(conn)
|
||||
if not query_success:
|
||||
logger.error("❌ 再次查询未完成的回调请求失败，停止任务执行")
|
||||
raise Exception("查询未完成的回调请求失败，任务停止执行")
|
||||
logger.info("3")
|
||||
if not callback_log_data:
|
||||
logger.info("📋 没有找到未完成的回调请求")
|
||||
return {"status": "skipped", "message": "没有找到未完成的回调请求"}
|
||||
|
||||
callback_log_id, site_id, request_headers_json, request_body_json = callback_log_data
|
||||
logger.info(f"📋 获取到未完成的回调请求: ID={callback_log_id}, site_id={site_id}")
|
||||
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 25, 'total': 100, 'status': f'获取回调请求成功(ID={callback_log_id}, site_id={site_id})，开始分析处理...'}
|
||||
)
|
||||
|
||||
# 解析请求头和请求体
|
||||
request_headers = json.loads(request_headers_json)
|
||||
request_body = json.loads(request_body_json)
|
||||
|
||||
# 保存callback_data.data中的数据
|
||||
data_list = request_body.get('data', [])
|
||||
if data_list and len(data_list) > 0:
|
||||
save_success = save_callback_data_items(conn, data_list, callback_log_id)
|
||||
if not save_success:
|
||||
logger.error(f"❌ 保存callback_data:{callback_log_id}失败，停止任务执行")
|
||||
raise Exception(f"保存callback_data:{callback_log_id}失败，任务停止执行")
|
||||
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 50, 'total': 100, 'status': f'请求体中通话明细处理成功(ID={callback_log_id}, site_id={site_id})，开始过滤需要转发的通话记录...'}
|
||||
)
|
||||
|
||||
# 根据手机号、task_id、user_id给data_list去重
|
||||
data_list = request_body.get('data', [])
|
||||
unique_data_list = []
|
||||
seen_records = set()
|
||||
original_count = len(data_list)
|
||||
|
||||
for item in data_list:
|
||||
# 提取手机号
|
||||
number_data = item.get('number_data', {})
|
||||
phone_number = number_data.get('number', '')
|
||||
|
||||
# 提取task_id
|
||||
task = item.get('task', {})
|
||||
task_id = task.get('id', '')
|
||||
|
||||
# 提取user_id
|
||||
user_id = item.get('user_id', '')
|
||||
|
||||
# 创建唯一标识
|
||||
unique_key = (phone_number, task_id, user_id)
|
||||
|
||||
# 如果这个组合没见过，则添加到去重列表中
|
||||
if unique_key not in seen_records:
|
||||
seen_records.add(unique_key)
|
||||
unique_data_list.append(item)
|
||||
|
||||
logger.info(f"🔄 数据去重完成: 原始数据 {original_count} 条，去重后 {len(unique_data_list)} 条 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
|
||||
if len(unique_data_list) < original_count:
|
||||
logger.info(f"🗑️ 移除了 {original_count - len(unique_data_list)} 条重复数据 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
|
||||
# 查询相关数据：根据手机号、task_id、user_id作为条件，按创建时间排序获取前三条记录
|
||||
query_success, related_records = get_related_records_by_unique_data_list(
|
||||
conn, unique_data_list, callback_log_id, settings.count_threshold
|
||||
)
|
||||
|
||||
if not query_success:
|
||||
logger.error("❌ 查询相关记录失败，停止任务执行")
|
||||
raise Exception("查询相关记录失败，任务停止执行")
|
||||
|
||||
# 判断是否有相关记录需要处理
|
||||
if not related_records:
|
||||
logger.info(f"📋 没有有效的通过记录需要处理，直接返回 - callback_log_id: {callback_log_id}, site_id={site_id}")
|
||||
return {"status": "completed", "message": "没有有效的通过记录需要处理"}
|
||||
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 70, 'total': 100, 'status': f'需要推送的通过记录已获取成功(ID={callback_log_id}, site_id={site_id})，准备转发...'}
|
||||
)
|
||||
|
||||
if related_records:
|
||||
# 创建只包含有效数据项的请求体
|
||||
filtered_request_body = request_body.copy()
|
||||
filtered_request_body['data'] = related_records
|
||||
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 85, 'total': 100, 'status': f'开始推送数据给DTC(ID={callback_log_id}, site_id={site_id})...'}
|
||||
)
|
||||
|
||||
success, retry_count = call_external_api_with_retry(
|
||||
conn=conn,
|
||||
request_body=filtered_request_body,
|
||||
request_headers=request_headers,
|
||||
max_retries=settings.external_api_retry_max,
|
||||
callback_failure_log_id=callback_log_id
|
||||
)
|
||||
|
||||
if success:
|
||||
logger.info(f"✅ 推送数据给DTC成功，处理的数据项数量: {len(related_records)}, 重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
else:
|
||||
logger.error(f"❌ 推送数据给DTC失败，重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
else:
|
||||
logger.info(f"📋 没有有效的通过记录需要处理 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
|
||||
logger.info(f"🎉 推送数据给DTC处理完成 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
|
||||
# 标记CallbackFailureLog为已完成
|
||||
mark_callback_log_completed(conn, callback_log_id)
|
||||
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 100, 'total': 100, 'status': f'标记回调请求日志为已完成(ID={callback_log_id}, site_id={site_id})'}
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "completed",
|
||||
"message": "任务完成"
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"❌ 任务执行失败: {e}", exc_info=True)
|
||||
return {"status": "error", "message": str(e)}
|
||||
|
||||
finally:
|
||||
# 释放分布式锁
|
||||
try:
|
||||
if 'lock' in locals():
|
||||
lock.release()
|
||||
logger.info(f"🔓 释放任务 {task_name} 的分布式锁")
|
||||
# 连接异步Redis
|
||||
connection_success = await async_redis_manager.connect()
|
||||
if not connection_success:
|
||||
logger.error(f"❌ Redis管理器连接失败，任务停止执行")
|
||||
raise Exception("Redis连接失败")
|
||||
|
||||
redis_pool = await async_redis_manager.get_redis_pool()
|
||||
|
||||
# 使用异步分布式锁
|
||||
async with distributed_lock(redis_pool, f"celery_task:{task_name}", timeout=300) as lock_acquired:
|
||||
if not lock_acquired:
|
||||
logger.warning(f"⚠️ 任务 {task_name} 正在执行中，跳过本次执行")
|
||||
return {"status": "skipped", "message": f"任务 {task_name} 正在执行中，跳过本次执行"}
|
||||
|
||||
logger.info(f"🔒 成功获取任务 {task_name} 的分布式锁")
|
||||
|
||||
async with AsyncSessionLocal() as db:
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 0, 'total': 100, 'status': f'正在获取下一条回调请求日志...'}
|
||||
)
|
||||
logger.info("2")
|
||||
# 获取一条未完成的回调请求（按创建时间取最小值）
|
||||
query_success, callback_log_data = await get_uncompleted_callback_log(db)
|
||||
if not query_success:
|
||||
logger.error("❌ 查询未完成的回调请求失败，尝试再查一次")
|
||||
query_success, callback_log_data = await get_uncompleted_callback_log(db)
|
||||
if not query_success:
|
||||
logger.error("❌ 再次查询未完成的回调请求失败，停止任务执行")
|
||||
raise Exception("查询未完成的回调请求失败，任务停止执行")
|
||||
logger.info("3")
|
||||
if not callback_log_data:
|
||||
logger.info("📋 没有找到未完成的回调请求")
|
||||
return {"status": "skipped", "message": "没有找到未完成的回调请求"}
|
||||
|
||||
callback_log_id, site_id, request_headers_json, request_body_json = callback_log_data
|
||||
logger.info(f"📋 获取到未完成的回调请求: ID={callback_log_id}, site_id={site_id}")
|
||||
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 25, 'total': 100, 'status': f'获取回调请求成功(ID={callback_log_id}, site_id={site_id})，开始分析处理...'}
|
||||
)
|
||||
|
||||
# 解析请求头和请求体
|
||||
request_headers = json.loads(request_headers_json)
|
||||
request_body = json.loads(request_body_json)
|
||||
|
||||
# 保存callback_data.data中的数据
|
||||
data_list = request_body.get('data', [])
|
||||
if data_list and len(data_list) > 0:
|
||||
save_success = await save_callback_data_items(db, data_list, callback_log_id)
|
||||
if not save_success:
|
||||
logger.error(f"❌ 保存callback_data:{callback_log_id}失败，停止任务执行")
|
||||
raise Exception(f"保存callback_data:{callback_log_id}失败，任务停止执行")
|
||||
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 50, 'total': 100, 'status': f'请求体中通话明细处理成功(ID={callback_log_id}, site_id={site_id})，开始过滤需要转发的通话记录...'}
|
||||
)
|
||||
|
||||
# 根据手机号、task_id、user_id给data_list去重
|
||||
data_list = request_body.get('data', [])
|
||||
unique_data_list = []
|
||||
seen_records = set()
|
||||
original_count = len(data_list)
|
||||
|
||||
for item in data_list:
|
||||
# 提取手机号
|
||||
number_data = item.get('number_data', {})
|
||||
phone_number = number_data.get('number', '')
|
||||
|
||||
# 提取task_id
|
||||
task = item.get('task', {})
|
||||
task_id = task.get('id', '')
|
||||
|
||||
# 提取user_id
|
||||
user_id = item.get('user_id', '')
|
||||
|
||||
# 创建唯一标识
|
||||
unique_key = (phone_number, task_id, user_id)
|
||||
|
||||
# 如果这个组合没见过，则添加到去重列表中
|
||||
if unique_key not in seen_records:
|
||||
seen_records.add(unique_key)
|
||||
unique_data_list.append(item)
|
||||
|
||||
logger.info(f"🔄 数据去重完成: 原始数据 {original_count} 条，去重后 {len(unique_data_list)} 条 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
|
||||
if len(unique_data_list) < original_count:
|
||||
logger.info(f"🗑️ 移除了 {original_count - len(unique_data_list)} 条重复数据 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
|
||||
# 查询相关数据：根据手机号、task_id、user_id作为条件，按创建时间排序获取前三条记录
|
||||
query_success, related_records = await get_related_records_by_unique_data_list(
|
||||
db, unique_data_list, callback_log_id, settings.count_threshold
|
||||
)
|
||||
|
||||
if not query_success:
|
||||
logger.error("❌ 查询相关记录失败，停止任务执行")
|
||||
raise Exception("查询相关记录失败，任务停止执行")
|
||||
|
||||
# 判断是否有相关记录需要处理
|
||||
if not related_records:
|
||||
logger.info(f"📋 没有有效的通过记录需要处理，直接返回 - callback_log_id: {callback_log_id}, site_id={site_id}")
|
||||
return {"status": "completed", "message": "没有有效的通过记录需要处理"}
|
||||
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 70, 'total': 100, 'status': f'需要推送的通过记录已获取成功(ID={callback_log_id}, site_id={site_id})，准备转发...'}
|
||||
)
|
||||
|
||||
if related_records:
|
||||
# 创建只包含有效数据项的请求体
|
||||
filtered_request_body = request_body.copy()
|
||||
filtered_request_body['data'] = related_records
|
||||
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 85, 'total': 100, 'status': f'开始推送数据给DTC(ID={callback_log_id}, site_id={site_id})...'}
|
||||
)
|
||||
|
||||
success, retry_count = await call_external_api_with_retry(
|
||||
db=db,
|
||||
request_body=filtered_request_body,
|
||||
request_headers=request_headers,
|
||||
max_retries=settings.external_api_retry_max,
|
||||
callback_failure_log_id=callback_log_id
|
||||
)
|
||||
|
||||
if success:
|
||||
logger.info(f"✅ 推送数据给DTC成功，处理的数据项数量: {len(related_records)}, 重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
else:
|
||||
logger.error(f"❌ 推送数据给DTC失败，重试次数: {retry_count} - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
else:
|
||||
logger.info(f"📋 没有有效的通过记录需要处理 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
|
||||
logger.info(f"🎉 推送数据给DTC处理完成 - callback_log_id: {callback_log_id}, site_id: {site_id}")
|
||||
|
||||
# 标记CallbackFailureLog为已完成
|
||||
await mark_callback_log_completed(db, callback_log_id)
|
||||
|
||||
# 更新任务状态
|
||||
self.update_state(
|
||||
state='PROGRESS',
|
||||
meta={'current': 100, 'total': 100, 'status': f'标记回调请求日志为已完成(ID={callback_log_id}, site_id={site_id})'}
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "completed",
|
||||
"message": "任务完成"
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"❌ 释放任务 {task_name} 的分布式锁失败: {e}")
|
||||
logger.error(f"❌ 任务执行失败: {e}", exc_info=True)
|
||||
return {"status": "error", "message": str(e)}
|
||||
|
||||
finally:
|
||||
# 分布式锁会通过上下文管理器自动释放
|
||||
logger.debug(f"🔓 任务 {task_name} 的分布式锁已通过上下文管理器处理")
|
||||
|
||||
# 在同步的Celery任务中运行异步代码
|
||||
try:
|
||||
# 创建新的事件循环
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
return loop.run_until_complete(async_task())
|
||||
except Exception as e:
|
||||
logger.error(f"❌ 异步任务执行失败: {e}", exc_info=True)
|
||||
return {"status": "error", "message": str(e)}
|
||||
finally:
|
||||
# 清理事件循环
|
||||
try:
|
||||
if 'loop' in locals():
|
||||
loop.close()
|
||||
except:
|
||||
pass
|
||||
|
||||
|
||||
@celery_app.task(bind=True, name='call_api', max_retries=None)
|
||||
|
||||
Reference in New Issue
Block a user