程序优化

libs/libs_db.py

@@ -17,8 +17,7 @@ def get_task(task_type):
     try:
         with pymysqlAlias() as conn:
             cursor = conn.cursor()
-
-            sql = f'select task_key from tasks where status = 0 order by priority desc, id asc limit 1'
+            sql = f'select task_key from tasks where status = 0 order by priority desc, id asc limit 1 for update'
             # print(f'sql: {sql}')
             cursor.execute(sql)
             data = cursor.fetchone()
@@ -49,6 +48,9 @@ def add_task(data):
         
 #新增任务到分布式处理任务表中 {"task_type":key,"task_key":pid,"psid":psid}
 def add_task_distributed(data):
+    flag = isInTaskDistributed(data["task_key"])
+    if flag:
+        return "already"
     try:
         with pymysqlAlias() as conn:
             cursor = conn.cursor() 
@@ -66,8 +68,8 @@ def start_task(data):
             cursor = conn.cursor()
 
             hostname = socket.gethostname()
-            sql = f'update tasks set status = 1, hostname = "{hostname}", started_at = now(), updated_at = now() where task_key = "{data["task_key"]} and status = 0"'
-            # print(f'sql: {sql}')
+            sql = f'update tasks set status = 1, hostname = "{hostname}", started_at = now(), updated_at = now() where status = 0 and task_key = {data["task_key"]}'
+            print(f'开始任务sql: {sql}')
             cursor.execute(sql)
             conn.commit()
     except Exception as e:
@@ -158,4 +160,20 @@ def isStudioConfigDistribute(psid):
                 return True
     except Exception as e:
         print(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())} 执行change_to_new_make_psid()异常: {str(e)}")
+        return "error"
+
+#判断是否已经存在了
+def isInTaskDistributed(task_key):
+    try:
+        with pymysqlAlias() as conn:
+            cursor = conn.cursor()
+            sql = f'select count(*) from task_distributed where status = 0 and task_key = "{task_key}"'
+            cursor.execute(sql)
+            result = cursor.fetchone()
+            if result[0] == 0:
+                return False
+            else:
+                return True
+    except Exception as e:
+        print(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())} 执行isInTaskDistributed()异常: {str(e)}")
         return "error"