修改连接池大小

core/database.py

@@ -17,17 +17,19 @@ def create_sql_engine(config):
     # 配置连接池
     engine = create_engine(
         f'mysql+mysqlconnector://{user}:{passwd}@{host}/{db_name}?charset={charset}',
-        pool_size=30,  # 连接池大小
+        pool_size=50,  # 连接池大小
         max_overflow=10,  # 超过连接池大小后可以创建的最大连接数
         pool_timeout=30,  # 获取连接的超时时间，单位为秒
         pool_recycle=3600,  # 连接最大复用时间，超过这个时间将被关闭并重新创建连接
     )
     return engine
 
+
 def create_rag_db_engine():
     config = configs.get()['database']['rag']
     return create_sql_engine(config)
 
+
 # 创建数据库引擎
 engine = create_rag_db_engine()
 
@@ -93,8 +95,8 @@ class DBHelper:
             logger.error(f"删除失败: {e}")
             raise
 
-    def get_all(self, model, **filters):
-        """获取所有符合条件的实体对象，支持更复杂的查询条件"""
+    def get_all(self, model, limit=None, **filters):
+        """获取所有符合条件的实体对象，支持更复杂的查询条件，并可限制最大返回条数"""
         try:
             query = self.session.query(model)
 
@@ -113,8 +115,14 @@ class DBHelper:
             if actual_filters:
                 query = query.filter_by(**actual_filters)
 
+            # 如果传入了 limit 参数，则限制返回的最大条数
+            if limit is not None:
+                query = query.limit(limit)
+
+            # 执行查询
             entities = query.all()
             return entities
+
         except SQLAlchemyError as e:
             logger.error(f"查询失败: {e}")
             raise

utils/keywords_utils.py

@@ -76,7 +76,7 @@ class KeywordSummaryTask:
         print('process_texts_concurrently start')
         db_helper = DBHelper()
         while True:
-            content_chunks = db_helper.get_all(ContentChunks, chunk_status=2, keywords_status=0)
+            content_chunks = db_helper.get_all(ContentChunks, limit=200, chunk_status=2, keywords_status=0)
             if len(content_chunks) == 0:
                 logger.info('sleep')
                 print('sleep')