本文目录导读：
我来详细介绍Python中自定义业务同步触发条件的几种常用方法：
基于时间条件的触发
import time
from datetime import datetime, timedelta
import schedule
class BusinessSyncer:
def __init__(self):
self.last_sync_time = None
self.sync_interval = 300 # 默认5分钟
# 定时触发
def sync_by_time(self, interval_minutes=5):
"""按固定时间间隔同步"""
current_time = datetime.now()
if self.last_sync_time is None:
self.last_sync_time = current_time
return True
# 计算时间差
time_diff = (current_time - self.last_sync_time).total_seconds()
if time_diff >= interval_minutes * 60:
self.last_sync_time = current_time
return True
return False
# 指定时间点触发
def sync_at_specific_time(self):
"""在特定时间点同步"""
current_time = datetime.now()
sync_times = ["09:00", "12:00", "15:00", "18:00", "21:00"]
current_time_str = current_time.strftime("%H:%M")
if current_time_str in sync_times:
return True
return False
基于数据变更的触发
import hashlib
import json
import os
class DataChangeTrigger:
def __init__(self):
self.data_checksum = None
self.data_file = "data_checksum.txt"
def calculate_checksum(self, data):
"""计算数据的校验和"""
data_str = json.dumps(data, sort_keys=True)
return hashlib.md5(data_str.encode()).hexdigest()
def check_data_changed(self, new_data):
"""检查数据是否发生变化"""
new_checksum = self.calculate_checksum(new_data)
if self.data_checksum is None:
self.data_checksum = new_checksum
return True
if new_checksum != self.data_checksum:
self.data_checksum = new_checksum
return True
return False
def check_file_changed(self, filepath):
"""检查文件是否被修改"""
if not os.path.exists(filepath):
return False
# 检查文件的最后修改时间
current_mtime = os.path.getmtime(filepath)
if not hasattr(self, 'last_mtime'):
self.last_mtime = current_mtime
return True
if current_mtime > self.last_mtime:
self.last_mtime = current_mtime
return True
return False
基于业务规则的触发
class BusinessRuleTrigger:
def __init__(self):
self.thresholds = {
"order_count": 100,
"revenue": 10000,
"error_rate": 0.05
}
def check_business_rules(self, business_metrics):
"""检查业务规则是否触发同步"""
triggers = []
# 规则1：订单量达到阈值
if business_metrics.get("order_count", 0) >= self.thresholds["order_count"]:
triggers.append("order_threshold")
# 规则2：收入达到阈值
if business_metrics.get("revenue", 0) >= self.thresholds["revenue"]:
triggers.append("revenue_threshold")
# 规则3：错误率超过阈值
if business_metrics.get("error_rate", 0) >= self.thresholds["error_rate"]:
triggers.append("error_threshold")
# 规则4：组合条件
if (business_metrics.get("order_count", 0) >= 50 and
business_metrics.get("revenue", 0) >= 5000):
triggers.append("combined_condition")
return triggers
def check_state_machine(self, current_state, event):
"""基于状态机的触发条件"""
state_machine = {
"init": {"data_ready": "syncing"},
"syncing": {"success": "synced", "failed": "error"},
"synced": {"new_data": "pending", "timeout": "syncing"},
"error": {"retry": "syncing", "force_sync": "syncing"}
}
if current_state in state_machine and event in state_machine[current_state]:
next_state = state_machine[current_state][event]
return next_state == "syncing"
return False
综合触发管理器
from typing import List, Dict, Any, Callable
import threading
import logging
class TriggerManager:
def __init__(self):
self.logger = logging.getLogger(__name__)
self.triggers: List[Dict[str, Any]] = []
self.sync_function = None
def register_trigger(self, name: str, condition_func: Callable,
priority: int = 0, cooldown: int = 60):
"""注册触发条件"""
trigger = {
"name": name,
"condition": condition_func,
"priority": priority,
"cooldown": cooldown,
"last_triggered": 0
}
self.triggers.append(trigger)
self.triggers.sort(key=lambda x: x["priority"], reverse=True)
def evaluate_triggers(self, context: Dict[str, Any]) -> List[str]:
"""评估所有触发条件"""
triggered = []
current_time = time.time()
for trigger in self.triggers:
try:
# 检查冷却时间
if current_time - trigger["last_triggered"] < trigger["cooldown"]:
continue
# 执行条件判断
if trigger["condition"](context):
trigger["last_triggered"] = current_time
triggered.append(trigger["name"])
self.logger.info(f"Trigger {trigger['name']} activated")
except Exception as e:
self.logger.error(f"Error evaluating trigger {trigger['name']}: {e}")
return triggered
def set_sync_function(self, func: Callable):
"""设置同步函数"""
self.sync_function = func
def run_sync(self, context: Dict[str, Any]):
"""执行同步"""
if not self.sync_function:
self.logger.error("No sync function set")
return
triggered_triggers = self.evaluate_triggers(context)
if triggered_triggers:
try:
self.logger.info(f"Starting sync due to triggers: {triggered_triggers}")
self.sync_function(context)
except Exception as e:
self.logger.error(f"Sync failed: {e}")
使用示例
# 创建触发管理实例
trigger_manager = TriggerManager()
# 定义同步函数
def sync_data(context):
print(f"Syncing data with context: {context}")
# 实际同步逻辑...
# 设置同步函数
trigger_manager.set_sync_function(sync_data)
# 创建各种触发器
class CustomTriggers:
@staticmethod
def order_volume_trigger(context):
return context.get("order_count", 0) > 100
@staticmethod
def time_interval_trigger(context):
# 每30分钟触发一次
return datetime.now().minute % 30 == 0
@staticmethod
def database_change_trigger(context):
# 检查数据库是否有新记录
return context.get("db_changed", False)
@staticmethod
def error_rate_trigger(context):
# 错误率超过0.1%时触发
return context.get("error_rate", 0) > 0.001
# 注册触发器
trigger_manager.register_trigger(
"order_volume",
CustomTriggers.order_volume_trigger,
priority=10,
cooldown=300
)
trigger_manager.register_trigger(
"time_interval",
CustomTriggers.time_interval_trigger,
priority=5,
cooldown=1800
)
trigger_manager.register_trigger(
"db_change",
CustomTriggers.database_change_trigger,
priority=8,
cooldown=60
)
trigger_manager.register_trigger(
"error_rate",
CustomTriggers.error_rate_trigger,
priority=9,
cooldown=120
)
# 业务上下文
business_context = {
"order_count": 150,
"error_rate": 0.002,
"db_changed": True,
"timestamp": datetime.now()
}
# 执行同步检查
trigger_manager.run_sync(business_context)
高级特性：组合条件触发器
class CompositeTrigger:
def __init__(self):
self.conditions = {}
def and_condition(self, *conditions):
"""AND条件组合"""
def check(context):
return all(cond(context) for cond in conditions)
return check
def or_condition(self, *conditions):
"""OR条件组合"""
def check(context):
return any(cond(context) for cond in conditions)
return check
def not_condition(self, condition):
"""NOT条件"""
def check(context):
return not condition(context)
return check
def threshold_condition(self, key, threshold, operator="greater"):
"""阈值条件"""
def check(context):
value = context.get(key, 0)
if operator == "greater":
return value > threshold
elif operator == "less":
return value < threshold
elif operator == "equal":
return value == threshold
return False
return check
# 使用组合条件
composite = CompositeTrigger()
# 创建复杂条件
critical_sync_condition = composite.and_condition(
composite.threshold_condition("order_count", 50, "greater"),
composite.or_condition(
composite.threshold_condition("revenue", 10000, "greater"),
composite.threshold_condition("error_rate", 0.05, "greater"),
composite.not_condition(
composite.threshold_condition("system_load", 90, "less")
)
)
)
# 注册组合条件
trigger_manager.register_trigger(
"critical_sync",
critical_sync_condition,
priority=20,
cooldown=60
)
这样的设计提供了：
- 灵活的触发条件：支持时间、数据变化、业务规则等多种触发方式
- 可组合性：可以组合多个条件形成复杂的触发逻辑
- 可配置性：支持优先级、冷却时间等配置
- 可扩展性：易于添加新的触发条件类型
- 监控和日志：记录了触发过程和状态变化
