Python异步任务调度器实战：构建高性能后台任务处理系统 | Python高级编程

高级特性与优化

class TaskDependencyManager:
    """任务依赖关系管理器"""
    
    def __init__(self):
        self.dependencies = defaultdict(list)
        self.dependents = defaultdict(list)
    
    def add_dependency(self, task_id: str, depends_on: List[str]):
        """添加任务依赖"""
        self.dependencies[task_id].extend(depends_on)
        for dep_id in depends_on:
            self.dependents[dep_id].append(task_id)
    
    def can_execute(self, task_id: str, completed_tasks: Set[str]) -> bool:
        """检查任务是否可以执行"""
        return all(
            dep_id in completed_tasks 
            for dep_id in self.dependencies.get(task_id, [])
        )
    
    def get_blocked_tasks(self, completed_tasks: Set[str]) -> List[str]:
        """获取被阻塞的任务"""
        blocked = []
        for task_id, deps in self.dependencies.items():
            if not self.can_execute(task_id, completed_tasks):
                blocked.append(task_id)
        return blocked

class LoadBalancer:
    """智能负载均衡器"""
    
    @staticmethod
    def select_best_node(nodes_metrics: List[Dict]) -> str:
        """选择最佳节点执行任务"""
        scored_nodes = []
        
        for node in nodes_metrics:
            score = 0
            
            # CPU权重
            cpu_score = 100 - node['cpu_percent']
            score += cpu_score * 0.4
            
            # 内存权重
            mem_score = 100 - node['memory_percent']
            score += mem_score * 0.3
            
            # 队列长度权重
            queue_score = max(0, 100 - node['queue_size'])
            score += queue_score * 0.2
            
            # 网络延迟权重
            latency_score = max(0, 100 - min(node.get('latency_ms', 100), 100))
            score += latency_score * 0.1
            
            scored_nodes.append((score, node['node_id']))
        
        # 返回得分最高的节点
        return max(scored_nodes)[1]

完整示例：电商订单处理系统

# 电商订单处理任务定义
class OrderProcessingSystem:
    
    def __init__(self):
        self.scheduler = DistributedScheduler()
        self.setup_tasks()
    
    def setup_tasks(self):
        """设置订单处理任务"""
        
        @self.scheduler.register_task
        @async_task(max_retries=5)
        async def validate_order(order_data: Dict) -> Dict:
            """验证订单"""
            await asyncio.sleep(0.5)  # 模拟验证过程
            if not order_data.get('items'):
                raise ValueError("订单中没有商品")
            return {"valid": True, "order_id": order_data['id']}
        
        @self.scheduler.register_task
        @async_task(max_retries=3)
        async def process_payment(order_id: str, amount: float) -> Dict:
            """处理支付"""
            await asyncio.sleep(1)  # 模拟支付处理
            # 模拟支付失败率10%
            import random
            if random.random() < 0.1:
                raise ValueError("支付处理失败")
            return {"paid": True, "transaction_id": f"txn_{order_id}"}
        
        @self.scheduler.register_task
        @async_task(max_retries=2)
        async def update_inventory(order_id: str, items: List[Dict]):
            """更新库存"""
            await asyncio.sleep(0.8)
            # 模拟库存更新
            return {"inventory_updated": True}
        
        @self.scheduler.register_task
        @async_task(max_retries=3)
        async def send_confirmation_email(order_id: str, email: str):
            """发送确认邮件"""
            await asyncio.sleep(0.3)
            # 模拟邮件发送
            return {"email_sent": True}
    
    async def process_complete_order(self, order_data: Dict):
        """处理完整订单流程"""
        tasks = []
        
        # 创建任务链
        validate_task_id = await self.scheduler.schedule(
            'validate_order', order_data
        )
        
        payment_task_id = await self.scheduler.schedule(
            'process_payment', 
            order_data['id'], 
            order_data['total_amount']
        )
        
        inventory_task_id = await self.scheduler.schedule(
            'update_inventory',
            order_data['id'],
            order_data['items']
        )
        
        email_task_id = await self.scheduler.schedule(
            'send_confirmation_email',
            order_data['id'],
            order_data['customer_email']
        )
        
        # 等待所有任务完成
        while True:
            all_done = True
            for task_id in [validate_task_id, payment_task_id, 
                          inventory_task_id, email_task_id]:
                status = self.scheduler.get_task_status(task_id)
                if status.get('status') not in ['success', 'failed']:
                    all_done = False
                    break
            
            if all_done:
                break
            await asyncio.sleep(1)
        
        # 收集结果
        results = {}
        for task_id in [validate_task_id, payment_task_id, 
                       inventory_task_id, email_task_id]:
            status = self.scheduler.get_task_status(task_id)
            results[task_id] = status
        
        return results

# 使用示例
async def main():
    order_system = OrderProcessingSystem()
    
    # 启动调度器
    scheduler_task = asyncio.create_task(
        order_system.scheduler.process_tasks()
    )
    
    # 处理订单
    order_data = {
        'id': 'order_123',
        'customer_email': 'customer@example.com',
        'items': [{'product_id': 'prod_1', 'quantity': 2}],
        'total_amount': 199.99
    }
    
    results = await order_system.process_complete_order(order_data)
    print("订单处理结果:", results)
    
    # 关闭系统
    order_system.scheduler.worker_pool.shutdown()
    scheduler_task.cancel()

if __name__ == "__main__":
    asyncio.run(main())

这个完整的电商订单处理系统展示了如何在实际业务场景中应用我们的任务调度器。

性能测试与优化建议

总结

通过本文的实战教程，我们构建了一个功能完整、性能优异的Python异步任务调度系统。这个系统具有以下特点：

• 分布式架构：支持多节点部署，具备良好的水平扩展能力
• 智能调度：支持优先级、延迟执行和依赖关系管理
• 高可用性：内置重试机制、死信队列和故障恢复
• 实时监控：提供Web界面和API进行系统监控
• 易于集成：简洁的API设计，方便集成到现有系统

相比现有的解决方案，我们的系统更加轻量级、可定制化，特别适合对性能和控制力有较高要求的场景。你可以基于这个基础框架，根据具体业务需求进行扩展和优化。