如何编写分布式任务脚本

wen 实用脚本 2026-06-27 236

本文目录导读：

如何编写分布式任务脚本

分布式任务核心架构
主流方案对比
完整实现示例
高级特性和最佳实践
部署和运维建议

我来详细介绍如何编写分布式任务脚本，包括架构设计、技术选型和具体实现方案。

分布式任务核心架构

基本架构模式

┌─────────────┐     ┌──────────────┐
│  调度器     │────▶│  任务队列    │
└─────────────┘     └──────┬───────┘
                            │
              ┌─────────────┼─────────────┐
              │             │             │
        ┌─────▼────┐ ┌─────▼────┐ ┌─────▼────┐
        │ 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]

这套分布式任务脚本系统支持：