fix: implement RedisLock for task execution to prevent concurrent processing

--bug=1068941@tapd-62980211 --user=刘瑞斌 集群部署时，因mk做了负载而触发器没有加锁，会触发两次 https://www.tapd.cn/62980211/s/1916945

Author: liuruibin

apps/trigger/handler/impl/trigger/scheduled_trigger.py

@@ -1,6 +1,8 @@
 # coding=utf-8
+from celery_once import QueueOnce
 from django.db.models import QuerySet
 
+from common.utils.lock import RedisLock
 from common.utils.logger import maxkb_logger
 from ops import celery_app
 from trigger.handler.base_trigger import BaseTrigger
@@ -239,17 +241,22 @@ def execute(trigger, **kwargs):
             maxkb_logger.warning(f"unsupported task={trigger_task}")
             return
         source_type = trigger_task["source_type"]
-
-        if source_type == "APPLICATION":
-            from trigger.handler.impl.task.application_task import ApplicationTask
-
-            ApplicationTask().execute(trigger_task, **kwargs)
-        elif source_type == "TOOL":
-            from trigger.handler.impl.task.tool_task import ToolTask
-
-            ToolTask().execute(trigger_task, **kwargs)
-        else:
-            maxkb_logger.warning(f"unsupported source_type={source_type}, task_id={trigger_task['id']}")
+        rlock = RedisLock()
+        source_id = str(trigger_task["source_id"])
+        if rlock.try_lock(source_id, 30 * 30):
+            try:
+                if source_type == "APPLICATION":
+                    from trigger.handler.impl.task.application_task import ApplicationTask
+
+                    ApplicationTask().execute(trigger_task, **kwargs)
+                elif source_type == "TOOL":
+                    from trigger.handler.impl.task.tool_task import ToolTask
+
+                    ToolTask().execute(trigger_task, **kwargs)
+                else:
+                    maxkb_logger.warning(f"unsupported source_type={source_type}, task_id={trigger_task['id']}")
+            finally:
+                rlock.un_lock(source_id)
 
     def support(self, trigger, **kwargs):
         return trigger.get("trigger_type") == "SCHEDULED"