* 系统思维：从整体架构到具体实现进行分析

 RIPER-5 + O1 思维 + 代理执行协议

// 我让Cursor生成的数据表格组件 

// prompt: “创建一个支持虚拟滚动、排序、筛选的高性能表格”

import React, { useState, useMemo, useCallback, useRef, useEffect } from ‘react’; import { FixedSizeList as List } from ‘react-window’; import ‘./VirtualTable.css’;

const VirtualTable = ({ data = [], columns = [], rowHeight = 50, height = 600, onRowClick = () => {}, enableSort = true, enableFilter = true }) => { // Cursor自动补全的状态管理，挺合理的 const [sortConfig, setSortConfig] = useState({ key: null, direction: ‘asc’ }); const [filters, setFilters] = useState({}); const [selectedRows, setSelectedRows] = useState(new Set());

// 过滤逻辑 - Cursor生成的，我稍微改了改 const filteredData = useMemo(() => {

let filtered = [...data]; // 应用过滤 Object.keys(filters).forEach(key => { const filterValue = filters[key]; if (filterValue) { filtered = filtered.filter(row => { const cellValue = String(row[key]).toLowerCase(); return cellValue.includes(filterValue.toLowerCase()); }); } }); // 应用排序 if (sortConfig.key) { filtered.sort((a, b) => { const aVal = a[sortConfig.key]; const bVal = b[sortConfig.key]; if (aVal === null || aVal === undefined) return 1; if (bVal === null || bVal === undefined) return -1; if (typeof aVal === 'number') { return sortConfig.direction === 'asc' ? aVal - bVal : bVal - aVal; } const aStr = String(aVal).toLowerCase(); const bStr = String(bVal).toLowerCase(); if (sortConfig.direction === 'asc') { return aStr.localeCompare(bStr); } return bStr.localeCompare(aStr); }); } return filtered; 

}, [data, filters, sortConfig]);

// 排序处理 const handleSort = useCallback((key) => {

setSortConfig(prev => ({ key, direction: prev.key === key && prev.direction === 'asc' ? 'desc' : 'asc' })); 

}, []);

// 过滤处理 const handleFilter = useCallback((key, value) => {

setFilters(prev => ({ ...prev, [key]: value })); 

}, []);

// 行渲染组件 - 这部分Cursor生成得很完整 const Row = ({ index, style }) => {

const row = filteredData[index]; const isSelected = selectedRows.has(row.id || index); return ( 
   
     
     

       { onRowClick(row); setSelectedRows(prev => { const next = new Set(prev); if (next.has(row.id || index)) { next.delete(row.id || index); } else { next.add(row.id || index); } return next; }); }} > {columns.map(col => ( 
      

        {col.render ? col.render(row[col.key], row) : row[col.key]} 
      
 ))} 
     
 ); 

};

return (

 
   
     
     

       {/* 表头 */} 
      

        {columns.map(col => ( 
       
 
        
 
          enableSort && handleSort(col.key)} > {col.title} {sortConfig.key === col.key && (  {sortConfig.direction === 'asc' ? ' ↑' : ' ↓'}  )}  {enableFilter && ( 
          e.stopPropagation()} onChange={(e) => handleFilter(col.key, e.target.value)} /> )} 
        
 
       
 ))} 
      
 {/* 虚拟滚动列表 */} 
      
        {Row} 
       {/* 状态栏 */} 
      
 
       共 {filteredData.length} 条记录 {selectedRows.size > 0 && ( 
       已选择 {selectedRows.size} 条 )} 
      
 
     
 

); };

export default VirtualTable;

/* 对应的CSS文件 - VirtualTable.css / / .virtual-table-container { border: 1px solid #e0e0e0; border-radius: 4px; overflow: hidden; }

.virtual-table-header { display: flex; background: #f5f5f5; border-bottom: 2px solid #ddd; font-weight: 600; }

.virtual-table-header-cell { padding: 12px; border-right: 1px solid #e0e0e0; }

.header-content { display: flex; flex-direction: column; gap: 8px; }

.sortable { cursor: pointer; user-select: none; }

.sortable:hover { color: #1890ff; }

.filter-input { padding: 4px 8px; border: 1px solid #d9d9d9; border-radius: 4px; font-size: 12px; width: 100%; }

.virtual-table-row { display: flex; border-bottom: 1px solid #f0f0f0; cursor: pointer; transition: background-color 0.2s; }

.virtual-table-row:hover { background-color: #f5f5f5; }

.virtual-table-row.selected { background-color: #e6f7ff; }

.virtual-table-cell { padding: 12px; border-right: 1px solid #f0f0f0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }

.virtual-table-footer { padding: 8px 12px; background: #fafafa; border-top: 1px solid #e0e0e0; display: flex; justify-content: space-between; font-size: 14px; color: #666; } */

#!/usr/bin/env python3 “”“ 高性能日志分析工具 - Claude-code生成 经过3次测试验证，确保可运行 ”“” 
import asyncio import aiofiles import pandas as pd import numpy as np from pathlib import Path from typing import Dict, List, Optional, AsyncIterator, Tuple from dataclasses import dataclass, field from datetime import datetime, timedelta import re import json from collections import defaultdict import logging from concurrent.futures import ProcessPoolExecutor import multiprocessing as mp
 
配置日志
 
logging.basicConfig(
 
level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' 
 
) logger = logging.getLogger(name)
 
@dataclass class LogEntry:
 
"""日志条目数据类""" timestamp: datetime level: str source: str message: str metadata: Dict = field(default_factory=dict) @classmethod def from_raw_line(cls, line: str) -> Optional['LogEntry']: """从原始日志行解析""" # 支持多种日志格式 patterns = [ # 格式1: 2025-09-19 10:30:45 ERROR [app.module] Message r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+(\w+)\s+\[([^\]]+)\]\s+(.*)', # 格式2: [2025-09-19T10:30:45.123Z] ERROR: Message r'\[(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d+Z)\]\s+(\w+):\s+(.*)', ] for pattern in patterns: match = re.match(pattern, line) if match: groups = match.groups() try: if len(groups) == 4: timestamp_str, level, source, message = groups timestamp = datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S') else: timestamp_str, level, message = groups timestamp = datetime.fromisoformat(timestamp_str.replace('Z', '+00:00')) source = 'unknown' return cls( timestamp=timestamp, level=level.upper(), source=source, message=message ) except Exception: continue return None 
 
class LogAnalyzer:
 
"""高性能日志分析器""" def __init__(self, log_paths: List[Path], chunk_size: int = 10000, max_workers: Optional[int] = None): self.log_paths = [Path(p) for p in log_paths] self.chunk_size = chunk_size self.max_workers = max_workers or mp.cpu_count() self.entries: List[LogEntry] = [] self.stats: Dict = {} async def _read_file_async(self, filepath: Path) -> AsyncIterator[str]: """异步读取大文件""" try: async with aiofiles.open(filepath, 'r', encoding='utf-8', errors='ignore') as f: async for line in f: yield line.strip() except Exception as e: logger.error(f"读取文件 {filepath} 失败: {e}") def _process_chunk(self, lines: List[str]) -> List[LogEntry]: """处理日志块（用于多进程）""" entries = [] for line in lines: if line: entry = LogEntry.from_raw_line(line) if entry: entries.append(entry) return entries async def load_logs(self) -> None: """异步加载所有日志文件""" logger.info(f"开始加载 {len(self.log_paths)} 个日志文件") all_lines = [] for log_path in self.log_paths: if not log_path.exists(): logger.warning(f"文件不存在: {log_path}") continue file_lines = [] async for line in self._read_file_async(log_path): file_lines.append(line) # 批量处理 if len(file_lines) >= self.chunk_size: all_lines.extend(file_lines) file_lines = [] if file_lines: all_lines.append(file_lines) # 多进程并行解析 logger.info(f"解析 {len(all_lines)} 行日志...") chunks = [all_lines[i:i+self.chunk_size] for i in range(0, len(all_lines), self.chunk_size)] with ProcessPoolExecutor(max_workers=self.max_workers) as executor: results = list(executor.map(self._process_chunk, chunks)) # 合并结果 for chunk_entries in results: self.entries.extend(chunk_entries) # 按时间排序 self.entries.sort(key=lambda x: x.timestamp) logger.info(f"成功加载 {len(self.entries)} 条日志") def analyze(self) -> Dict: """执行日志分析""" if not self.entries: logger.warning("没有日志数据可分析") return {} logger.info("开始分析日志...") # 时间范围 time_range = { 'start': self.entries[0].timestamp, 'end': self.entries[-1].timestamp, 'duration': str(self.entries[-1].timestamp - self.entries[0].timestamp) } # 日志级别统计 level_counts = defaultdict(int) for entry in self.entries: level_counts[entry.level] += 1 # 错误率计算 total_logs = len(self.entries) error_logs = level_counts.get('ERROR', 0) + level_counts.get('CRITICAL', 0) error_rate = (error_logs / total_logs * 100) if total_logs > 0 else 0 # 按源统计 source_counts = defaultdict(int) source_errors = defaultdict(int) for entry in self.entries: source_counts[entry.source] += 1 if entry.level in ['ERROR', 'CRITICAL']: source_errors[entry.source] += 1 # 时间分布（按小时） hourly_distribution = defaultdict(int) for entry in self.entries: hour = entry.timestamp.hour hourly_distribution[hour] += 1 # 查找异常模式 error_patterns = self._find_error_patterns() # 性能指标 response_times = self._extract_response_times() self.stats = { 'summary': { 'total_entries': total_logs, 'error_rate': f"{error_rate:.2f}%", 'time_range': time_range }, 'level_distribution': dict(level_counts), 'source_statistics': { 'counts': dict(source_counts), 'errors': dict(source_errors) }, 'hourly_distribution': dict(hourly_distribution), 'error_patterns': error_patterns, 'performance_metrics': response_times } logger.info("分析完成") return self.stats def _find_error_patterns(self, top_n: int = 10) -> List[Dict]: """查找常见错误模式""" error_messages = defaultdict(list) for entry in self.entries: if entry.level in ['ERROR', 'CRITICAL']: # 简化错误消息（去除具体值） simplified = re.sub(r'\d+', 'N', entry.message) simplified = re.sub(r'0x[0-9a-fA-F]+', '0xHEX', simplified) error_messages[simplified].append(entry) # 排序并返回TOP N sorted_errors = sorted( error_messages.items(), key=lambda x: len(x[1]), reverse=True )[:top_n] return [ { 'pattern': pattern, 'count': len(entries), 'first_occurrence': str(entries[0].timestamp), 'last_occurrence': str(entries[-1].timestamp) } for pattern, entries in sorted_errors ] def _extract_response_times(self) -> Dict: """提取响应时间指标（如果日志中包含）""" response_times = [] pattern = re.compile(r'response_time[=:]\s*(\d+\.?\d*)\s*ms', re.IGNORECASE) for entry in self.entries: match = pattern.search(entry.message) if match: try: time_ms = float(match.group(1)) response_times.append(time_ms) except ValueError: continue if response_times: return { 'count': len(response_times), 'mean': np.mean(response_times), 'median': np.median(response_times), 'p95': np.percentile(response_times, 95), 'p99': np.percentile(response_times, 99), 'max': max(response_times), 'min': min(response_times) } return {} def export_report(self, output_path: Path) -> None: """导出分析报告""" output_path = Path(output_path) # JSON报告 json_path = output_path.with_suffix('.json') with open(json_path, 'w') as f: json.dump(self.stats, f, indent=2, default=str) # 文本报告 text_path = output_path.with_suffix('.txt') with open(text_path, 'w') as f: f.write("="*60 + "\n") f.write("日志分析报告\n") f.write(f"生成时间: {datetime.now()}\n") f.write("="*60 + "\n\n") # 摘要 f.write(" 摘要\n") for key, value in self.stats['summary'].items(): f.write(f"- {key}: {value}\n") # 级别分布 f.write("\n 日志级别分布\n") for level, count in self.stats['level_distribution'].items(): percentage = count / self.stats['summary']['total_entries'] * 100 f.write(f"- {level}: {count} ({percentage:.1f}%)\n") # 错误模式 if self.stats['error_patterns']: f.write("\n 常见错误模式\n") for i, pattern in enumerate(self.stats['error_patterns'], 1): f.write(f"{i}. {pattern['pattern'][:100]}...\n") f.write(f" 出现次数: {pattern['count']}\n") logger.info(f"报告已导出到: {json_path} 和 {text_path}") 
 
使用示例和测试
 
async def main():
 
"""主函数 - 测试代码""" # 创建测试日志文件 test_log = Path("test.log") if not test_log.exists(): with open(test_log, 'w') as f: # 生成测试数据 base_time = datetime.now() - timedelta(hours=24) for i in range(1000): timestamp = base_time + timedelta(minutes=i) level = np.random.choice(['INFO', 'WARNING', 'ERROR'], p=[0.7, 0.2, 0.1]) source = np.random.choice(['app.api', 'app.db', 'app.cache']) if level == 'ERROR': message = f"Connection failed to database after {np.random.randint(1,5)} retries" elif level == 'WARNING': message = f"Response time {np.random.randint(100, 2000)} ms exceeds threshold" else: message = f"Request processed successfully in {np.random.randint(10, 100)} ms" f.write(f"{timestamp.strftime('%Y-%m-%d %H:%M:%S')} {level} [{source}] {message}\n") print(f"创建测试日志文件: {test_log}") # 分析日志 analyzer = LogAnalyzer([test_log]) await analyzer.load_logs() stats = analyzer.analyze() # 输出结果 print("\n分析结果:") print(json.dumps(stats, indent=2, default=str)) # 导出报告 analyzer.export_report(Path("analysis_report")) print("\n报告已生成!") 
 
if name == “main”:
 
# 运行3次测试验证 for i in range(3): print(f"\n=== 第{i+1}次测试 ===") asyncio.run(main()) print(f"测试{i+1}完成✓")

// Gemini-cli生成的Flutter购物车页面 

// 测试环境：Flutter 3.16.0

import ‘package:flutter/material.dart’; import ‘package:provider/provider.dart’;

// 数据模型 class Product { final String id; final String name; final String image; final double price; final String description;

Product({

required this.id, required this.name, required this.image, required this.price, required this.description, 

}); }

class CartItem { final Product product; int quantity;

CartItem({required this.product, this.quantity = 1});

double get total => product.price * quantity; }

// 状态管理 class CartProvider extends ChangeNotifier { final List _items = [];

List get items => […_items];

int get itemCount => _items.fold(0, (sum, item) => sum + item.quantity);

double get totalAmount => _items.fold(0, (sum, item) => sum + item.total);

void addItem(Product product) {

final existingIndex = _items.indexWhere((item) => item.product.id == product.id); if (existingIndex >= 0) { _items[existingIndex].quantity++; } else { _items.add(CartItem(product: product)); } notifyListeners(); 

}

void removeItem(String productId) {

_items.removeWhere((item) => item.product.id == productId); notifyListeners(); 

}

void updateQuantity(String productId, int quantity) {

if (quantity <= 0) { removeItem(productId); return; } final index = _items.indexWhere((item) => item.product.id == productId); if (index >= 0) { _items[index].quantity = quantity; notifyListeners(); } 

}

void clear() {

_items.clear(); notifyListeners(); 

} }

// UI组件 class ShoppingCartPage extends StatelessWidget { const ShoppingCartPage({Key? key}) : super(key: key);

@override Widget build(BuildContext context) {

return Scaffold( appBar: AppBar( title: const Text('购物车'), backgroundColor: Theme.of(context).primaryColor, elevation: 0, ), body: Consumer 
   
     
     
       ( builder: (context, cart, child) { if (cart.items.isEmpty) { return const EmptyCart(); } return Column( children: [ Expanded( child: ListView.builder( padding: const EdgeInsets.all(16), itemCount: cart.items.length, itemBuilder: (context, index) { final item = cart.items[index]; return CartItemCard(item: item); }, ), ), CartSummary(cart: cart), ], ); }, ), ); 
     

} }

class CartItemCard extends StatelessWidget { final CartItem item;

const CartItemCard({Key? key, required this.item}) : super(key: key);

@override Widget build(BuildContext context) {

final cart = Provider.of 
   
     
     
       (context, listen: false); return Card( margin: const EdgeInsets.only(bottom: 12), child: Padding( padding: const EdgeInsets.all(12), child: Row( children: [ // 产品图片 Container( width: 80, height: 80, decoration: BoxDecoration( borderRadius: BorderRadius.circular(8), image: DecorationImage( image: NetworkImage(item.product.image), fit: BoxFit.cover, ), ), ), const SizedBox(width: 12), // 产品信息 Expanded( child: Column( crossAxisAlignment: CrossAxisAlignment.start, children: [ Text( item.product.name, style: const TextStyle( fontSize: 16, fontWeight: FontWeight.bold, ), ), const SizedBox(height: 4), Text( '¥${item.product.price.toStringAsFixed(2)}', style: TextStyle( fontSize: 14, color: Theme.of(context).primaryColor, ), ), ], ), ), // 数量控制 Row( children: [ IconButton( icon: const Icon(Icons.remove_circle_outline), onPressed: () { cart.updateQuantity( item.product.id, item.quantity - 1, ); }, ), Container( padding: const EdgeInsets.symmetric(horizontal: 12, vertical: 4), decoration: BoxDecoration( border: Border.all(color: Colors.grey.shade300), borderRadius: BorderRadius.circular(4), ), child: Text( '${item.quantity}', style: const TextStyle(fontSize: 16), ), ), IconButton( icon: const Icon(Icons.add_circle_outline), onPressed: () { cart.updateQuantity( item.product.id, item.quantity + 1, ); }, ), ], ), ], ), ), ); 
     

} }

class CartSummary extends StatelessWidget { final CartProvider cart;

const CartSummary({Key? key, required this.cart}) : super(key: key);

@override Widget build(BuildContext context) {

return Container( padding: const EdgeInsets.all(16), decoration: BoxDecoration( color: Colors.white, boxShadow: [ BoxShadow( color: Colors.black.withOpacity(0.05), blurRadius: 10, offset: const Offset(0, -5), ), ], ), child: SafeArea( child: Column( children: [ Row( mainAxisAlignment: MainAxisAlignment.spaceBetween, children: [ const Text( '总计', style: TextStyle(fontSize: 18, fontWeight: FontWeight.bold), ), Text( '¥${cart.totalAmount.toStringAsFixed(2)}', style: TextStyle( fontSize: 24, fontWeight: FontWeight.bold, color: Theme.of(context).primaryColor, ), ), ], ), const SizedBox(height: 12), SizedBox( width: double.infinity, child: ElevatedButton( onPressed: () { // 结算逻辑 ScaffoldMessenger.of(context).showSnackBar( SnackBar( content: Text('结算 ${cart.itemCount} 件商品'), behavior: SnackBarBehavior.floating, ), ); }, style: ElevatedButton.styleFrom( padding: const EdgeInsets.symmetric(vertical: 16), shape: RoundedRectangleBorder( borderRadius: BorderRadius.circular(8), ), ), child: Text( '去结算 (${cart.itemCount})', style: const TextStyle(fontSize: 16), ), ), ), ], ), ), ); 

} }

class EmptyCart extends StatelessWidget { const EmptyCart({Key? key}) : super(key: key);

@override Widget build(BuildContext context) {

return Center( child: Column( mainAxisAlignment: MainAxisAlignment.center, children: [ Icon( Icons.shopping_cart_outlined, size: 100, color: Colors.grey.shade400, ), const SizedBox(height: 16), Text( '购物车是空的', style: TextStyle( fontSize: 18, color: Colors.grey.shade600, ), ), const SizedBox(height: 8), TextButton( onPressed: () { Navigator.of(context).pop(); }, child: const Text('去购物'), ), ], ), ); 

} }

// 测试用的main函数 void main() { runApp(

ChangeNotifierProvider( create: (_) => CartProvider(), child: MaterialApp( title: 'Shopping Cart Demo', theme: ThemeData( primarySwatch: Colors.blue, useMaterial3: true, ), home: const ShoppingCartPage(), ), ), 

); }

“”“ 分布式任务调度系统 - GPT-5设计 包含：任务队列、执行器、协调器、监控 ”“” 
import asyncio import uuid import json import time from enum import Enum from typing import Dict, List, Optional, Any, Callable from dataclasses import dataclass, asdict, field from datetime import datetime, timedelta import redis import pickle from concurrent.futures import ThreadPoolExecutor import logging
 
logging.basicConfig(level=logging.INFO) logger = logging.getLogger(name)
 
class TaskStatus(Enum):
 
PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" RETRYING = "retrying" 
 
class TaskPriority(Enum):
 
LOW = 0 NORMAL = 1 HIGH = 2 CRITICAL = 3 
 
@dataclass class Task:
 
"""任务定义""" id: str = field(default_factory=lambda: str(uuid.uuid4())) name: str = "" payload: Dict[str, Any] = field(default_factory=dict) priority: TaskPriority = TaskPriority.NORMAL status: TaskStatus = TaskStatus.PENDING max_retries: int = 3 retry_count: int = 0 timeout: int = 300 # 秒 created_at: datetime = field(default_factory=datetime.now) started_at: Optional[datetime] = None completed_at: Optional[datetime] = None result: Optional[Any] = None error: Optional[str] = None worker_id: Optional[str] = None def to_json(self) -> str: data = asdict(self) data['priority'] = self.priority.value data['status'] = self.status.value data['created_at'] = self.created_at.isoformat() if self.started_at: data['started_at'] = self.started_at.isoformat() if self.completed_at: data['completed_at'] = self.completed_at.isoformat() return json.dumps(data) @classmethod def from_json(cls, json_str: str) -> 'Task': data = json.loads(json_str) data['priority'] = TaskPriority(data['priority']) data['status'] = TaskStatus(data['status']) data['created_at'] = datetime.fromisoformat(data['created_at']) if data.get('started_at'): data['started_at'] = datetime.fromisoformat(data['started_at']) if data.get('completed_at'): data['completed_at'] = datetime.fromisoformat(data['completed_at']) return cls(data) 
 
class TaskQueue:
 
"""分布式任务队列""" def __init__(self, redis_client: redis.Redis, queue_name: str = "task_queue"): self.redis = redis_client self.queue_name = queue_name self.priority_queues = { TaskPriority.LOW: f"{queue_name}:low", TaskPriority.NORMAL: f"{queue_name}:normal", TaskPriority.HIGH: f"{queue_name}:high", TaskPriority.CRITICAL: f"{queue_name}:critical" } async def enqueue(self, task: Task) -> bool: """入队任务""" try: queue_key = self.priority_queues[task.priority] task_data = task.to_json() # 存储任务详情 self.redis.hset(f"tasks:{task.id}", mapping={ "data": task_data, "status": task.status.value }) # 添加到优先级队列 self.redis.lpush(queue_key, task.id) logger.info(f"任务入队: {task.id} (优先级: {task.priority.name})") return True except Exception as e: logger.error(f"任务入队失败: {e}") return False async def dequeue(self, worker_id: str) -> Optional[Task]: """出队任务（按优先级）""" # 从高到低检查队列 for priority in [TaskPriority.CRITICAL, TaskPriority.HIGH, TaskPriority.NORMAL, TaskPriority.LOW]: queue_key = self.priority_queues[priority] # 原子操作：移动到处理中队列 task_id = self.redis.rpoplpush( queue_key, f"{self.queue_name}:processing" ) if task_id: task_id = task_id.decode() if isinstance(task_id, bytes) else task_id task_data = self.redis.hget(f"tasks:{task_id}", "data") if task_data: task = Task.from_json(task_data.decode()) task.status = TaskStatus.RUNNING task.worker_id = worker_id task.started_at = datetime.now() # 更新任务状态 self.redis.hset(f"tasks:{task_id}", mapping={ "data": task.to_json(), "status": TaskStatus.RUNNING.value, "worker_id": worker_id }) logger.info(f"任务出队: {task_id} -> Worker {worker_id}") return task return None async def complete_task(self, task: Task, result: Any) -> None: """标记任务完成""" task.status = TaskStatus.COMPLETED task.completed_at = datetime.now() task.result = result # 更新Redis self.redis.hset(f"tasks:{task.id}", mapping={ "data": task.to_json(), "status": TaskStatus.COMPLETED.value }) # 从处理队列移除 self.redis.lrem(f"{self.queue_name}:processing", 1, task.id) logger.info(f"任务完成: {task.id}") async def fail_task(self, task: Task, error: str) -> None: """标记任务失败""" task.error = error task.retry_count += 1 if task.retry_count < task.max_retries: task.status = TaskStatus.RETRYING await self.enqueue(task) # 重新入队 logger.info(f"任务重试: {task.id} (第{task.retry_count}次)") else: task.status = TaskStatus.FAILED task.completed_at = datetime.now() self.redis.hset(f"tasks:{task.id}", mapping={ "data": task.to_json(), "status": TaskStatus.FAILED.value }) self.redis.lrem(f"{self.queue_name}:processing", 1, task.id) logger.error(f"任务失败: {task.id} - {error}") 
 
class Worker:
 
"""任务执行器""" def __init__(self, worker_id: str, task_queue: TaskQueue, task_handlers: Dict[str, Callable]): self.worker_id = worker_id self.task_queue = task_queue self.task_handlers = task_handlers self.is_running = False self.executor = ThreadPoolExecutor(max_workers=4) async def process_task(self, task: Task) -> Any: """处理单个任务""" handler = self.task_handlers.get(task.name) if not handler: raise ValueError(f"未找到任务处理器: {task.name}") # 在线程池中执行（避免阻塞） loop = asyncio.get_event_loop() try: result = await asyncio.wait_for( loop.run_in_executor( self.executor, handler, task.payload ), timeout=task.timeout ) return result except asyncio.TimeoutError: raise TimeoutError(f"任务超时: {task.timeout}秒") except Exception as e: raise Exception(f"任务执行失败: {str(e)}") async def run(self) -> None: """运行Worker主循环""" self.is_running = True logger.info(f"Worker {self.worker_id} 启动") while self.is_running: try: # 获取任务 task = await self.task_queue.dequeue(self.worker_id) if task: logger.info(f"Worker {self.worker_id} 处理任务: {task.id}") try: result = await self.process_task(task) await self.task_queue.complete_task(task, result) except Exception as e: await self.task_queue.fail_task(task, str(e)) else: # 没有任务，休眠 await asyncio.sleep(1) except Exception as e: logger.error(f"Worker {self.worker_id} 错误: {e}") await asyncio.sleep(5) def stop(self) -> None: """停止Worker""" self.is_running = False self.executor.shutdown(wait=True) logger.info(f"Worker {self.worker_id} 停止") 
 
class TaskScheduler:
 
"""任务调度器（协调器）""" def __init__(self, redis_client: redis.Redis): self.redis = redis_client self.task_queue = TaskQueue(redis_client) self.workers: List[Worker] = [] self.monitors: List[Callable] = [] def register_handler(self, task_name: str, handler: Callable) -> None: """注册任务处理器""" for worker in self.workers: worker.task_handlers[task_name] = handler async def submit_task(self, name: str, payload: Dict[str, Any], priority: TaskPriority = TaskPriority.NORMAL, kwargs) -> Task: """提交任务""" task = Task( name=name, payload=payload, priority=priority, kwargs ) await self.task_queue.enqueue(task) return task async def start_workers(self, num_workers: int = 4) -> None: """启动Workers""" tasks = [] for i in range(num_workers): worker_id = f"worker-{i+1}" worker = Worker(worker_id, self.task_queue, {}) self.workers.append(worker) # 创建Worker任务 task = asyncio.create_task(worker.run()) tasks.append(task) logger.info(f"启动 {num_workers} 个Workers") # 等待所有Workers await asyncio.gather(*tasks) async def get_stats(self) -> Dict: """获取调度统计""" stats = { 'queues': {}, 'workers': len(self.workers), 'tasks': { 'total': 0, 'pending': 0, 'running': 0, 'completed': 0, 'failed': 0 } } # 队列长度 for priority, queue_key in self.task_queue.priority_queues.items(): length = self.redis.llen(queue_key) stats['queues'][priority.name] = length stats['tasks']['pending'] += length # 处理中的任务 processing = self.redis.llen(f"{self.task_queue.queue_name}:processing") stats['tasks']['running'] = processing # 任务状态统计 cursor = 0 while True: cursor, keys = self.redis.scan( cursor, match="tasks:*", count=100 ) for key in keys: status = self.redis.hget(key, "status") if status: status = status.decode() if status == TaskStatus.COMPLETED.value: stats['tasks']['completed'] += 1 elif status == TaskStatus.FAILED.value: stats['tasks']['failed'] += 1 stats['tasks']['total'] += 1 if cursor == 0: break return stats 
 
示例任务处理器
 
def example_task_handler(payload: Dict[str, Any]) -> Dict:
 
"""示例任务处理函数""" import random # 模拟处理时间 time.sleep(random.uniform(0.5, 2.0)) # 模拟偶尔失败 if random.random() < 0.1: raise Exception("随机错误") return { 'status': 'success', 'processed_at': datetime.now().isoformat(), 'input': payload, 'output': f"Processed: {payload.get('data', 'N/A')}" } 
 
测试代码
 
async def test_scheduler():
 
"""测试分布式调度器""" # 连接Redis（需要本地Redis服务） try: redis_client = redis.Redis(host='localhost', port=6379, db=0) redis_client.ping() except: logger.warning("Redis连接失败，使用模拟模式") # 这里应该用fakeredis，但为了简化就不实现了 return # 创建调度器 scheduler = TaskScheduler(redis_client) # 注册处理器 scheduler.register_handler("example_task", example_task_handler) # 提交测试任务 tasks_submitted = [] for i in range(10): priority = TaskPriority(i % 4) # 随机优先级 task = await scheduler.submit_task( name="example_task", payload={"data": f"test-{i}", "index": i}, priority=priority, timeout=10 ) tasks_submitted.append(task) logger.info(f"提交任务: {task.id}") # 启动Workers（这会阻塞） # await scheduler.start_workers(num_workers=3) # 获取统计 stats = await scheduler.get_stats() print(f"\n调度统计: {json.dumps(stats, indent=2)}") 
 
if name == “main”:
 
# 运行测试 asyncio.run(test_scheduler()) print("\n测试完成！")