From 379b730e96bb834b4196e2eb4ff0f2f3cc9f4535 Mon Sep 17 00:00:00 2001 From: Tanshin Pan Date: Mon, 18 May 2026 10:17:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(core):=20=E6=B7=BB=E5=8A=A0=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BA=93=E4=BC=9A=E8=AF=9D=E7=AE=A1=E7=90=86=E5=92=8C?= =?UTF-8?q?Redis=E9=98=9F=E5=88=97=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在article API中添加session关闭确保资源释放 - 为Db类的所有数据库操作方法添加session管理的finally块 - 修复add_article方法中的异常处理逻辑，确保回滚时session存在 - 优化TaskQueueManager中的Redis存储策略，从全量重写改为增量追加 - 统一所有数据库查询方法使用相同的session管理模式 --- apis/article.py | 2 ++ core/db.py | 43 +++++++++++++++++++++++++++++++++++-------- core/queue/queue.py | 13 ++++++++++--- 3 files changed, 47 insertions(+), 11 deletions(-) diff --git a/apis/article.py b/apis/article.py index 6ac02563..6cd1e6df 100644 --- a/apis/article.py +++ b/apis/article.py @@ -167,6 +167,8 @@ async def clean_orphan_articles( message="清理无效文章失败" ) ) + finally: + session.close() @router.delete("/clean-old", summary="清理指定天数前的旧文章") diff --git a/core/db.py b/core/db.py index 6f4c05d7..d357fb84 100644 --- a/core/db.py +++ b/core/db.py @@ -123,6 +123,7 @@ class Db: def __exit__(self, exc_type, exc_val, exc_tb): self.close() def delete_article(self,article_data:dict)->bool: + session = None try: art = Article(**article_data) if art.id: # type: ignore @@ -136,9 +137,13 @@ class Db: except Exception as e: print_error(f"delete article:{str(e)}") pass + finally: + if session is not None: + session.close() return False def add_article(self, article_data: dict,check_exist=True) -> bool: + session = None try: session=self.get_session() from datetime import datetime @@ -194,48 +199,70 @@ class Db: session.add(art) print_info(f"Added article: {art.id}") sta=session.commit() + return True except Exception as e: - session.rollback() # 回滚事务，确保session状态正常 + if session: + session.rollback() # 回滚事务，确保session状态正常 if "UNIQUE" in str(e) or "Duplicate entry" in str(e): print_warning(f"Article already exists: {art.id}") else: print_error(f"Failed to add article: {e}") return False - return True - + finally: + if session is not None: + session.close() def get_articles(self, id:str=None, limit:int=30, offset:int=0) -> List[Article]: # type: ignore + session = None try: - data = self.get_session().query(Article).limit(limit).offset(offset) + session = self.get_session() + data = session.query(Article).limit(limit).offset(offset) return data except Exception as e: print(f"Failed to fetch Feed: {e}") return e # type: ignore + finally: + if session is not None: + session.close() def get_all_mps(self) -> List[Feed]: """Get all Feed records""" + session = None try: - return self.get_session().query(Feed).all() + session = self.get_session() + return session.query(Feed).filter(Feed.status == 1).all() except Exception as e: print(f"Failed to fetch Feed: {e}") return e # type: ignore + finally: + if session is not None: + session.close() def get_mps_list(self, mp_ids:str) -> List[Feed]: + session = None try: ids=mp_ids.split(',') - data = self.get_session().query(Feed).filter(Feed.id.in_(ids)).all() + session = self.get_session() + data = session.query(Feed).filter(Feed.id.in_(ids)).all() return data except Exception as e: print(f"Failed to fetch Feed: {e}") return e # type: ignore + finally: + if session is not None: + session.close() def get_mps(self, mp_id:str) -> Optional[Feed]: + session = None try: ids=mp_id.split(',') - data = self.get_session().query(Feed).filter_by(id= mp_id).first() + session = self.get_session() + data = session.query(Feed).filter_by(id= mp_id).first() return data except Exception as e: print(f"Failed to fetch Feed: {e}") return e # type: ignore - + finally: + if session is not None: + session.close() def get_faker_id(self, mp_id:str): data = self.get_mps(mp_id) return data.faker_id # type: ignore diff --git a/core/queue/queue.py b/core/queue/queue.py index d2fb2a03..35f199f3 100644 --- a/core/queue/queue.py +++ b/core/queue/queue.py @@ -368,9 +368,16 @@ class TaskQueueManager: max_retries=max_retries )) - # 保存到 Redis - self._save_pending_to_redis() - self._save_status_to_redis() + # 优化：只追加新任务到 Redis，而不是全量重写 + redis_client = _get_redis() + if redis_client: + try: + new_item = self._pending_items[-1].to_dict() + redis_client.rpush(self._redis_keys['pending'], json.dumps(new_item, ensure_ascii=False)) + # 只更新计数，不保存完整状态 + redis_client.hincrby(self._redis_keys['status'], 'pending_count', 1) + except Exception as e: + print_error(f"追加任务到 Redis 失败: {e}") # 标记需要广播，但不在这里执行（避免在锁内进行异步操作） broadcast_needed = True