本文目录导读：
我来详细介绍如何编写分布式任务脚本，包括架构设计、技术选型和具体实现方案。
分布式任务核心架构
基本架构模式
┌─────────────┐ ┌──────────────┐
│ 调度器 │────▶│ 任务队列 │
└─────────────┘ └──────┬───────┘
│
┌─────────────┼─────────────┐
│ │ │
┌─────▼────┐ ┌─────▼────┐ ┌─────▼────┐
│ Worker 1 │ │ Worker 2 │ │ Worker 3 │
└──────────┘ └──────────┘ └──────────┘主流方案对比
# 方案1: Celery (最成熟) # 方案2: Redis Queue (轻量级) # 方案3: Apache Airflow (DAG工作流) # 方案4: 自建RabbitMQ + 工作进程
完整实现示例
基础分布式架构实现
# scheduler.py - 任务调度器
import redis
import json
import uuid
from typing import Callable, Dict, Any
import pickle
class DistributedTaskScheduler:
def __init__(self, redis_host='localhost', redis_port=6379):
self.redis_client = redis.Redis(host=redis_host, port=redis_port)
self.task_queue = 'task_queue'
self.result_backend = 'task_results'
def submit_task(self, func_path: str, *args, **kwargs) -> str:
"""提交分布式任务"""
task_id = str(uuid.uuid4())
task = {
'id': task_id,
'func_path': func_path,
'args': args,
'kwargs': kwargs,
'status': 'pending'
}
# 序列化并推送任务到队列
self.redis_client.rpush(self.task_queue, json.dumps(task))
return task_id
def get_task_result(self, task_id: str) -> Dict:
"""获取任务执行结果"""
result = self.redis_client.get(f'{self.result_backend}:{task_id}')
return json.loads(result) if result else None
工作节点实现
# worker.py - 任务执行器
import redis
import json
import importlib
import time
from typing import Dict, Any
class TaskWorker:
def __init__(self, worker_id: str, redis_host='localhost', redis_port=6379):
self.worker_id = worker_id
self.redis_client = redis.Redis(host=redis_host, port=redis_port)
self.task_queue = 'task_queue'
self.result_backend = 'task_results'
def _load_function(self, func_path: str):
"""动态加载任务函数"""
module_path, func_name = func_path.rsplit('.', 1)
module = importlib.import_module(module_path)
return getattr(module, func_name)
def _execute_task(self, task: Dict[str, Any]) -> Dict:
"""执行单个任务"""
try:
func = self._load_function(task['func_path'])
result = func(*task['args'], **task['kwargs'])
return {
'status': 'completed',
'result': result,
'worker_id': self.worker_id
}
except Exception as e:
return {
'status': 'failed',
'error': str(e),
'worker_id': self.worker_id
}
def start_working(self):
"""开始持续处理任务"""
print(f"Worker {self.worker_id} started")
while True:
# 阻塞获取任务（超时10秒）
task_data = self.redis_client.blpop(self.task_queue, timeout=10)
if task_data:
task = json.loads(task_data[1])
print(f"Processing task: {task['id']}")
# 执行任务
result = self._execute_task(task)
# 保存结果
result_key = f'{self.result_backend}:{task["id"]}'
self.redis_client.set(result_key, json.dumps(result))
time.sleep(0.1)
任务函数定义
# tasks.py - 具体的分布式任务
import time
import random
def compute_pi(iterations: int) -> float:
"""计算π的近似值（耗时任务示例）"""
pi = 0
for i in range(iterations):
pi += 4 * ((-1) ** i) / (2 * i + 1)
time.sleep(0.01) # 模拟计算延迟
return pi
def process_image(image_path: str, filters: list) -> dict:
"""图像处理任务示例"""
# 模拟图像处理
time.sleep(random.uniform(1, 3))
return {
'image_path': image_path,
'filters_applied': filters,
'processed': True
}
def batch_data_analysis(data_chunk: list) -> dict:
"""数据分析任务示例"""
return {
'count': len(data_chunk),
'sum': sum(data_chunk),
'mean': sum(data_chunk) / len(data_chunk) if data_chunk else 0
}
使用Celery实现（生产环境推荐）
# celery_app.py
from celery import Celery
from celery.signals import task_failure, task_success
app = Celery('distributed_tasks',
broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1')
# Celery配置
app.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='Asia/Shanghai',
enable_utc=True,
worker_max_tasks_per_child=100, # 每个worker最大任务数
task_acks_late=True, # 任务确认延迟
worker_prefetch_multiplier=1, # 预取因子
)
# 定义任务
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def long_running_task(self, data: list):
try:
# 耗时计算
result = process_data(data)
return result
except Exception as exc:
raise self.retry(exc=exc)
@app.task(bind=True, rate_limit='10/m') # 限速
def rate_limited_task(self, item: str):
return f"Processed: {item}"
使用Redis实现负载均衡
# balanced_worker.py - 带负载均衡的工作节点
import redis
import json
import psutil
import time
class BalancedWorker:
def __init__(self, worker_id: str, max_load: float = 0.8):
self.worker_id = worker_id
self.max_load = max_load
self.redis_client = redis.Redis()
self.current_tasks = 0
self.max_concurrent_tasks = 5
def get_system_load(self) -> float:
"""获取系统负载"""
return psutil.cpu_percent() / 100.0
def can_accept_task(self) -> bool:
"""判断是否可以接受新任务"""
return (self.current_tasks < self.max_concurrent_tasks and
self.get_system_load() < self.max_load)
def register_worker(self):
"""注册worker到服务发现"""
worker_info = {
'id': self.worker_id,
'load': self.get_system_load(),
'capacity': self.max_concurrent_tasks - self.current_tasks,
'last_heartbeat': time.time()
}
self.redis_client.hset('workers', self.worker_id, json.dumps(worker_info))
def heartbeat(self):
"""定期发送心跳"""
while True:
self.register_worker()
time.sleep(5)
监控和状态管理
# monitor.py - 任务监控器
import redis
import json
from datetime import datetime
from typing import Dict, List
class TaskMonitor:
def __init__(self, redis_host='localhost'):
self.redis_client = redis.Redis(host=redis_host)
def get_queue_status(self) -> Dict:
"""获取队列状态"""
queue_length = self.redis_client.llen('task_queue')
workers = self.redis_client.hgetall('workers')
return {
'pending_tasks': queue_length,
'active_workers': len(workers),
'workers': {
k.decode(): json.loads(v.decode())
for k, v in workers.items()
}
}
def get_task_statistics(self) -> Dict:
"""获取任务统计信息"""
completed = 0
failed = 0
# 扫描结果后端
for key in self.redis_client.scan_iter('task_results:*'):
result = json.loads(self.redis_client.get(key))
if result['status'] == 'completed':
completed += 1
elif result['status'] == 'failed':
failed += 1
return {
'completed': completed,
'failed': failed,
'total': completed + failed
}
启动和管理脚本
# manage.py - 分布式任务管理系统
import subprocess
import sys
import time
from multiprocessing import Process
def start_worker(worker_id: str, count: int = 3):
"""启动多个worker"""
processes = []
for i in range(count):
p = Process(target=run_worker, args=(f'{worker_id}-{i}',))
p.start()
processes.append(p)
return processes
def run_worker(worker_id: str):
"""运行单个worker"""
from worker import TaskWorker
worker = TaskWorker(worker_id)
worker.start_working()
def submit_test_task():
"""提交测试任务"""
from scheduler import DistributedTaskScheduler
scheduler = DistributedTaskScheduler()
# 提交多个计算任务
task_ids = []
for i in range(10):
task_id = scheduler.submit_task(
'tasks.compute_pi',
iterations=1000 + i * 500
)
task_ids.append(task_id)
print(f"Submitted task {task_id}")
return task_ids
if __name__ == '__main__':
if len(sys.argv) < 2:
print("Usage: python manage.py [start|stop|status|test]")
sys.exit(1)
command = sys.argv[1]
if command == 'start':
print("Starting distributed task system...")
start_worker('worker', count=3)
elif command == 'test':
print("Submitting test tasks...")
submit_test_task()
高级特性和最佳实践
任务优先级处理
# priority_queue.py
class PriorityTaskScheduler:
def __init__(self):
self.redis_client = redis.Redis()
self.queues = {
'high': 'task_queue:high',
'medium': 'task_queue:medium',
'low': 'task_queue:low'
}
def submit_with_priority(self, task_func, priority='medium', **kwargs):
queue = self.queues.get(priority, self.queues['medium'])
task = {
'id': str(uuid.uuid4()),
'func': task_func,
'args': kwargs.get('args', ()),
'kwargs': kwargs.get('kwargs', {}),
'priority': priority
}
self.redis_client.rpush(queue, json.dumps(task))
任务重试和容错
# retry_handler.py
class TaskRetryHandler:
MAX_RETRIES = 3
RETRY_DELAYS = [10, 30, 60] # 指数退避
def handle_failed_task(self, task: Dict, error: str) -> bool:
"""处理失败任务"""
retries = task.get('retries', 0)
if retries < self.MAX_RETRIES:
# 延迟重试
delay = self.RETRY_DELAYS[retries]
task['retries'] = retries + 1
task['next_retry'] = time.time() + delay
# 重新入队
self.redis_client.zadd('delayed_tasks',
{json.dumps(task): task['next_retry']})
return True
# 最大重试次数后移入死信队列
self.redis_client.rpush('dead_letter_queue', json.dumps(task))
return False
部署和运维建议
-
使用容器化部署：
# Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["python", "manage.py", "start"]
-
监控和告警配置：
# 集成Prometheus指标 from prometheus_client import Counter, Histogram, start_http_server
task_count = Counter('tasks_processed_total', 'Number of processed tasks') task_duration = Histogram('task_duration_seconds', 'Task duration in seconds')
3. **配置管理**：
```yaml
# config.yaml
redis:
host: localhost
port: 6379
password:
db: 0
worker:
max_concurrent: 5
max_retries: 3
heartbeat_interval: 10
task_queue:
max_size: 10000
priority_levels: [high, medium, low]这套分布式任务脚本系统支持：
- 任务队列管理和调度
- 多worker并行执行
- 负载均衡和健康检查
- 任务优先级处理
- 重试和容错机制
- 监控和统计功能
可以根据具体需求选择合适的实现方案和配置参数。
