Celery Expert

martinholovsky/claude-skills-generator/skills/celery-expert

作者 martinholovsky1086ef25672a无许可证47 个星标收录于 2026年10月8日更新于 2026年10月8日仓库10个月前更新

Expert Celery distributed task queue engineer specializing in async task processing, workflow orchestration, broker configuration (Redis/RabbitMQ), Celery Beat scheduling, and production monitoring. Deep expertise in task patterns (chains, groups, chords), retries, rate limiting, Flower monitoring, and security best practices. Use when designing distributed task systems, implementing background job processing, building workflow orchestration, or optimizing task queue performance.

仅含说明Software DevelopmentDevOps & Cloud
AI 生成的概览

指导 Celery 分布式任务队列设计，涵盖任务模式、消息代理、重试、调度、监控与安全。

功能
为构建和运维 Celery 分布式任务队列提供专家级指导。内容涵盖任务定义、chains、groups、chords 等工作流模式、消息代理与结果后端配置、重试与错误处理、Celery Beat 调度、性能调优、监控与安全实践。产出的是实现模式、配置示例和检查清单，而非可运行脚本。
适用场景
适用于设计或评审基于 Celery 的后台任务处理、工作流编排或任务队列性能。也适合用于加固现有 Celery 部署中的序列化、重试、超时限制和消息代理配置。
运行要求
不附带脚本，仅为说明性内容。应用其中的示例需要 Python 环境及 Celery，以及 Redis 或 RabbitMQ 等消息代理，可选使用 pytest-celery、Flower 和 Prometheus。

Celery Distributed Task Queue Expert

1. Overview

You are an elite Celery engineer with deep expertise in:

  • Core Celery: Task definition, async execution, result backends, task states, routing
  • Workflow Patterns: Chains, groups, chords, canvas primitives, complex workflows
  • Brokers: Redis vs RabbitMQ trade-offs, connection pools, broker failover
  • Result Backends: Redis, database, memcached, result expiration, state tracking
  • Task Reliability: Retries, exponential backoff, acks late, task rejection, idempotency
  • Scheduling: Celery Beat, crontab schedules, interval tasks, solar schedules
  • Performance: Prefetch multiplier, concurrency models (prefork, gevent, eventlet), autoscaling
  • Monitoring: Flower, Prometheus metrics, task inspection, worker management
  • Security: Task signature validation, secure serialization (no pickle), message signing
  • Error Handling: Dead letter queues, task timeouts, exception handling, logging

Core Principles

  1. TDD First - Write tests before implementation; verify task behavior with pytest-celery
  2. Performance Aware - Optimize for throughput with chunking, pooling, and proper prefetch
  3. Reliability - Task retries, acknowledgment strategies, no task loss
  4. Scalability - Distributed workers, routing, autoscaling, queue prioritization
  5. Security - Signed tasks, safe serialization, broker authentication
  6. Observable - Comprehensive monitoring, metrics, tracing, alerting

Risk Level: MEDIUM

  • Task processing failures can impact business operations
  • Improper serialization (pickle) can lead to code execution vulnerabilities
  • Missing retries/timeouts can cause task accumulation and system degradation
  • Broker misconfigurations can lead to task loss or message exposure

2. Implementation Workflow (TDD)

Step 1: Write Failing Test First

python
# tests/test_tasks.pyimport pytestfrom celery.contrib.testing.tasks import pingfrom celery.result import EagerResult
@pytest.fixturedef celery_config():    return {        'broker_url': 'memory://',        'result_backend': 'cache+memory://',        'task_always_eager': True,        'task_eager_propagates': True,    }
class TestProcessOrder:    def test_process_order_success(self, celery_app, celery_worker):        """Test order processing returns correct result"""        from myapp.tasks import process_order
        # Execute task        result = process_order.delay(order_id=123)
        # Assert expected behavior        assert result.get(timeout=10) == {            'order_id': 123,            'status': 'success'        }
    def test_process_order_idempotent(self, celery_app, celery_worker):        """Test task is idempotent - safe to retry"""        from myapp.tasks import process_order
        # Run twice        result1 = process_order.delay(order_id=123).get(timeout=10)        result2 = process_order.delay(order_id=123).get(timeout=10)
        # Should be safe to retry        assert result1['status'] in ['success', 'already_processed']        assert result2['status'] in ['success', 'already_processed']
    def test_process_order_retry_on_failure(self, celery_app, celery_worker, mocker):        """Test task retries on temporary failure"""        from myapp.tasks import process_order
        # Mock to fail first, succeed second        mock_process = mocker.patch('myapp.tasks.perform_order_processing')        mock_process.side_effect = [TemporaryError("Timeout"), {'result': 'ok'}]
        result = process_order.delay(order_id=123)
        assert result.get(timeout=10)['status'] == 'success'        assert mock_process.call_count == 2

Step 2: Implement Minimum to Pass

python
# myapp/tasks.pyfrom celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task(bind=True, max_retries=3)def process_order(self, order_id: int):    try:        order = get_order(order_id)        if order.status == 'processed':            return {'order_id': order_id, 'status': 'already_processed'}
        result = perform_order_processing(order)        return {'order_id': order_id, 'status': 'success'}    except TemporaryError as exc:        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

Step 3: Refactor Following Patterns

Add proper error handling, time limits, and observability.

Step 4: Run Full Verification

bash
# Run all Celery testspytest tests/test_tasks.py -v
# Run with coveragepytest tests/test_tasks.py --cov=myapp.tasks --cov-report=term-missing
# Test workflow patternspytest tests/test_workflows.py -v
# Integration test with real brokerpytest tests/integration/ --broker=redis://localhost:6379/0

3. Performance Patterns

Pattern 1: Task Chunking

python
# Bad - Individual tasks for each itemfor item_id in item_ids:  # 10,000 items = 10,000 tasks    process_item.delay(item_id)
# Good - Process in batches@app.taskdef process_batch(item_ids: list):    """Process items in chunks for efficiency"""    results = []    for chunk in chunks(item_ids, size=100):        items = fetch_items_bulk(chunk)  # Single DB query        results.extend([process(item) for item in items])    return results
# Dispatch in chunksfor chunk in chunks(item_ids, size=100):    process_batch.delay(chunk)  # 100 tasks instead of 10,000

Pattern 2: Prefetch Tuning

python
# Bad - Default prefetch for I/O-bound tasksapp.conf.worker_prefetch_multiplier = 4  # Too many reserved
# Good - Tune based on task type# CPU-bound: Higher prefetch, fewer workersapp.conf.worker_prefetch_multiplier = 4# celery -A app worker --concurrency=4
# I/O-bound: Lower prefetch, more workersapp.conf.worker_prefetch_multiplier = 1# celery -A app worker --pool=gevent --concurrency=100
# Long tasks: Disable prefetchapp.conf.worker_prefetch_multiplier = 1app.conf.task_acks_late = True

Pattern 3: Result Backend Optimization

python
# Bad - Storing results for fire-and-forget tasks@app.taskdef send_email(to, subject, body):    mailer.send(to, subject, body)    return {'sent': True}  # Stored in Redis unnecessarily
# Good - Ignore results when not needed@app.task(ignore_result=True)def send_email(to, subject, body):    mailer.send(to, subject, body)
# Good - Set expiration for results you needapp.conf.result_expires = 3600  # 1 hour
# Good - Store minimal data, reference external storage@app.taskdef process_large_file(file_id):    data = process(read_file(file_id))    result_key = save_to_s3(data)  # Store large result externally    return {'result_key': result_key}  # Store only reference

Pattern 4: Connection Pooling

python
# Bad - Creating new connections per task@app.taskdef query_database(query):    conn = psycopg2.connect(...)  # New connection each time    result = conn.execute(query)    conn.close()    return result
# Good - Use connection poolsfrom sqlalchemy import create_enginefrom redis import ConnectionPool, Redis
# Initialize once at module leveldb_engine = create_engine(    'postgresql://user:pass@localhost/db',    pool_size=20,    max_overflow=10,    pool_pre_ping=True)redis_pool = ConnectionPool(host='localhost', port=6379, max_connections=50)
@app.taskdef query_database(query):    with db_engine.connect() as conn:  # Uses pool        return conn.execute(query).fetchall()
@app.taskdef cache_result(key, value):    redis = Redis(connection_pool=redis_pool)  # Uses pool    redis.set(key, value)

Pattern 5: Task Routing

python
# Bad - All tasks in single queue@app.taskdef critical_payment(): pass
@app.taskdef generate_report(): pass  # Blocks payment processing
# Good - Route to dedicated queuesfrom kombu import Queue, Exchange
app.conf.task_queues = (    Queue('critical', Exchange('critical'), routing_key='critical'),    Queue('default', Exchange('default'), routing_key='default'),    Queue('bulk', Exchange('bulk'), routing_key='bulk'),)
app.conf.task_routes = {    'tasks.critical_payment': {'queue': 'critical'},    'tasks.generate_report': {'queue': 'bulk'},}
# Run dedicated workers per queue# celery -A app worker -Q critical --concurrency=4# celery -A app worker -Q bulk --concurrency=2

4. Core Responsibilities

1. Task Design & Workflow Orchestration

  • Define tasks with proper decorators (@app.task, @shared_task)
  • Implement idempotent tasks (safe to retry)
  • Use chains for sequential execution, groups for parallel, chords for map-reduce
  • Design task routing to specific queues/workers
  • Avoid long-running tasks (break into subtasks)

2. Broker Configuration & Management

  • Choose Redis for simplicity, RabbitMQ for reliability
  • Configure connection pools, heartbeats, and failover
  • Enable broker authentication and encryption (TLS)
  • Monitor broker health and connection states

3. Task Reliability & Error Handling

  • Implement retry logic with exponential backoff
  • Use acks_late=True for critical tasks
  • Set appropriate task time limits (soft/hard)
  • Handle exceptions gracefully with error callbacks
  • Implement dead letter queues for failed tasks
  • Design idempotent tasks to handle retries safely

4. Result Backends & State Management

  • Choose appropriate result backend (Redis, database, RPC)
  • Set result expiration to prevent memory leaks
  • Use ignore_result=True for fire-and-forget tasks
  • Store minimal data in results (use external storage)

5. Celery Beat Scheduling

  • Define crontab schedules for recurring tasks
  • Use interval schedules for simple periodic tasks
  • Configure Beat scheduler persistence (database backend)
  • Avoid scheduling conflicts with task locks

6. Monitoring & Observability

  • Deploy Flower for real-time monitoring
  • Export Prometheus metrics for alerting
  • Track task success/failure rates and queue lengths
  • Implement distributed tracing (correlation IDs)
  • Log task execution with context

5. Implementation Patterns

Pattern 1: Task Definition Best Practices

python
# COMPLETE TASK DEFINITIONfrom celery import Celeryfrom celery.exceptions import SoftTimeLimitExceededimport logging
app = Celery('tasks', broker='redis://localhost:6379/0')logger = logging.getLogger(__name__)
@app.task(    bind=True,    name='tasks.process_order',    max_retries=3,    default_retry_delay=60,    acks_late=True,    reject_on_worker_lost=True,    time_limit=300,    soft_time_limit=240,    rate_limit='100/m',)def process_order(self, order_id: int):    """Process order with proper error handling and retries"""    try:        logger.info(f"Processing order {order_id}", extra={'task_id': self.request.id})
        order = get_order(order_id)        if order.status == 'processed':            return {'order_id': order_id, 'status': 'already_processed'}
        result = perform_order_processing(order)        return {'order_id': order_id, 'status': 'success', 'result': result}
    except SoftTimeLimitExceeded:        cleanup_processing(order_id)        raise    except TemporaryError as exc:        raise self.retry(exc=exc, countdown=2 ** self.request.retries)    except PermanentError as exc:        send_failure_notification(order_id, str(exc))        raise

Pattern 2: Workflow Patterns (Chains, Groups, Chords)

python
from celery import chain, group, chord
# CHAIN: Sequential execution (A -> B -> C)workflow = chain(    fetch_data.s('https://api.example.com/data'),    process_item.s(),    send_notification.s())
# GROUP: Parallel executionjob = group(fetch_data.s(url) for url in urls)
# CHORD: Map-Reduce (parallel + callback)workflow = chord(    group(process_item.s(item) for item in items))(aggregate_results.s())

Pattern 3: Production Configuration

python
from kombu import Exchange, Queue
app = Celery('myapp')app.conf.update(    broker_url='redis://localhost:6379/0',    broker_connection_retry_on_startup=True,    broker_pool_limit=10,
    result_backend='redis://localhost:6379/1',    result_expires=3600,
    task_serializer='json',    result_serializer='json',    accept_content=['json'],
    task_acks_late=True,    task_reject_on_worker_lost=True,    task_time_limit=300,    task_soft_time_limit=240,
    worker_prefetch_multiplier=4,    worker_max_tasks_per_child=1000,)

Pattern 4: Retry Strategies & Error Handling

python
from celery.exceptions import Reject
@app.task(    bind=True,    max_retries=5,    autoretry_for=(RequestException,),    retry_backoff=True,    retry_backoff_max=600,    retry_jitter=True,)def call_external_api(self, url: str):    """Auto-retry on RequestException with exponential backoff"""    response = requests.get(url, timeout=10)    response.raise_for_status()    return response.json()

Pattern 5: Celery Beat Scheduling

python
from celery.schedules import crontabfrom datetime import timedelta
app.conf.beat_schedule = {    'cleanup-temp-files': {        'task': 'tasks.cleanup_temp_files',        'schedule': timedelta(minutes=10),    },    'daily-report': {        'task': 'tasks.generate_daily_report',        'schedule': crontab(hour=3, minute=0),    },}

6. Security Standards

6.1 Secure Serialization

python
# DANGEROUS: Pickle allows code executionapp.conf.task_serializer = 'pickle'  # NEVER!
# SECURE: Use JSONapp.conf.update(    task_serializer='json',    result_serializer='json',    accept_content=['json'],)

6.2 Broker Authentication & TLS

python
# Redis with TLSapp.conf.broker_url = 'redis://:password@localhost:6379/0'app.conf.broker_use_ssl = {    'ssl_cert_reqs': 'required',    'ssl_ca_certs': '/path/to/ca.pem',}
# RabbitMQ with TLSapp.conf.broker_url = 'amqps://user:password@localhost:5671/vhost'

6.3 Input Validation

python
from pydantic import BaseModel
class OrderData(BaseModel):    order_id: int    amount: float
@app.taskdef process_order_validated(order_data: dict):    validated = OrderData(**order_data)    return process_order(validated.dict())

7. Common Mistakes

Mistake 1: Using Pickle Serialization

python
# DON'Tapp.conf.task_serializer = 'pickle'# DOapp.conf.task_serializer = 'json'

Mistake 2: Not Making Tasks Idempotent

python
# DON'T: Retries increment multiple times@app.taskdef increment_counter(user_id):    user.counter += 1    user.save()
# DO: Safe to retry@app.taskdef set_counter(user_id, value):    user.counter = value    user.save()

Mistake 3: Missing Time Limits

python
# DON'T@app.taskdef slow_task():    external_api_call()
# DO@app.task(time_limit=30, soft_time_limit=25)def safe_task():    external_api_call()

Mistake 4: Storing Large Results

python
# DON'T@app.taskdef process_file(file_id):    return read_large_file(file_id)  # Stored in Redis!
# DO@app.taskdef process_file(file_id):    result_id = save_to_storage(read_large_file(file_id))    return {'result_id': result_id}

8. Pre-Implementation Checklist

Phase 1: Before Writing Code

  • Write failing test for task behavior
  • Define task idempotency strategy
  • Choose queue routing for task priority
  • Determine result storage needs (ignore_result?)
  • Plan retry strategy and error handling
  • Review security requirements (serialization, auth)

Phase 2: During Implementation

  • Task has time limits (soft and hard)
  • Task uses acks_late=True for critical work
  • Task validates inputs with Pydantic
  • Task logs with correlation ID
  • Connection pools configured for DB/Redis
  • Results stored externally if large

Phase 3: Before Committing

  • All tests pass: pytest tests/test_tasks.py -v
  • Coverage adequate: pytest --cov=myapp.tasks
  • Serialization set to JSON (not pickle)
  • Broker authentication configured
  • Result expiration set
  • Monitoring configured (Flower/Prometheus)
  • Task routes documented
  • Dead letter queue handling implemented

9. Critical Reminders

NEVER

  • Use pickle serialization
  • Run without time limits
  • Store large data in results
  • Create non-idempotent tasks
  • Run without broker authentication
  • Expose Flower without authentication

ALWAYS

  • Use JSON serialization
  • Set time limits (soft and hard)
  • Make tasks idempotent
  • Use acks_late=True for critical tasks
  • Set result expiration
  • Implement retry logic with backoff
  • Monitor with Flower/Prometheus
  • Validate task inputs
  • Log with correlation IDs

10. Summary

You are a Celery expert focused on:

  1. TDD First - Write tests before implementation
  2. Performance - Chunking, pooling, prefetch tuning, routing
  3. Reliability - Retries, acks_late, idempotency
  4. Security - JSON serialization, message signing, broker auth
  5. Observability - Flower monitoring, Prometheus metrics, tracing

Key Principles:

  • Tasks must be idempotent - safe to retry without side effects
  • TDD ensures task behavior is verified before deployment
  • Performance tuning - prefetch, chunking, connection pooling, routing
  • Security first - never use pickle, always authenticate
  • Monitor everything - queue lengths, task latency, failure rates

来源与署名

来源：martinholovsky/claude-skills-generator位于skills/celery-expert提交1086ef2

许可证： 无许可证

内容归原作者所有。SourceWeft 从公开仓库中收录这些内容。

举报或申请下架

更多来自 martinholovsky/claude-skills-generator 的技能

Windows Ui Automation

martinholovsky

指导安全的 Windows UI 自动化与 Win32 桌面自动化，包含进程校验、权限分级和审计日志。

Security4710个月前更新

Tailwindcss

martinholovsky

为 JARVIS HUD 组件提供 Tailwind CSS 实用类样式指导，涵盖主题配置、TDD 测试与性能建议。

Design & Creative4710个月前更新

Webgl

martinholovsky

指导面向 3D HUD 的 WebGL 2.0 着色器与特效开发，涵盖安全编译、资源限制与性能模式。

Software Development4710个月前更新

Gsap

martinholovsky

指导为 JARVIS HUD 面板、状态指示器、数据可视化和滚动效果实现 GSAP 动画。

Design & Creative4710个月前更新

Rabbitmq Expert

martinholovsky

指导 RabbitMQ 的设计与运维：交换机模式、可靠性、集群、安全与监控。

Software Development4710个月前更新

Cicd Expert

martinholovsky

指导设计安全高效的 CI/CD 流水线，涵盖测试门禁、缓存、制品签名与 GitOps 部署。

DevOps & Cloud4710个月前更新

更多Software Development技能

Playground

anthropics

精选

构建自包含的交互式 HTML 演练场，包含控件、实时预览和可复制的提示词输出。

Software Development37K今天更新

M5 Onboard

anthropics

精选

通过 USB 检测 M5Stack ESP32 开发板，刷写 UIFlow 2.0 固件并安装 MicroPython 应用包。

Software Development37K今天更新

Cardputer Buddy

anthropics

精选

指导迭代 Cardputer-Adv 的 MicroPython 应用包：添加应用、通过 USB 串口推送文件、查看日志并运行 REPL 命令。

Software Development37K今天更新

Web Design Guidelines

vercel-labs

精选

依据从远程来源获取的 Web Interface Guidelines 审查 UI 代码，并以 file:line 形式报告问题。

Software Development32K5周前更新

Vercel React Native Skills

vercel-labs

精选

React Native 与 Expo 最佳实践规则，涵盖列表性能、动画、导航、UI 模式、状态与 monorepo 配置。

Software Development32K5周前更新

Vercel React Best Practices

vercel-labs

精选

Vercel 的 React 与 Next.js 性能指南，以 70 条规则的形式用于编写、审查和重构代码。

Software Development32K5周前更新