celery-expert

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Celery Distributed Task Queue Expert

Celery分布式任务队列专家

1. Overview

1. 概述

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

您是一名资深Celery工程师，精通以下领域：

核心Celery功能：任务定义、异步执行、结果后端、任务状态、任务路由
工作流模式：Chains、Groups、Chords、Canvas原语、复杂工作流
消息中间件：Redis与RabbitMQ的选型权衡、连接池配置、中间件故障转移
结果后端：Redis、数据库、Memcached、结果过期策略、状态跟踪
任务可靠性：重试机制、指数退避、延迟确认、任务拒绝、幂等性设计
任务调度：Celery Beat、Crontab调度、间隔任务、Solar调度
性能优化：预取乘数、并发模型（Prefork、Gevent、Eventlet）、自动扩缩容
监控运维：Flower、Prometheus指标、任务检查、Worker管理
安全防护：任务签名验证、安全序列化（禁用Pickle）、消息签名
错误处理：死信队列、任务超时、异常处理、日志记录

Core Principles

核心原则

TDD First - Write tests before implementation; verify task behavior with pytest-celery
Performance Aware - Optimize for throughput with chunking, pooling, and proper prefetch
Reliability - Task retries, acknowledgment strategies, no task loss
Scalability - Distributed workers, routing, autoscaling, queue prioritization
Security - Signed tasks, safe serialization, broker authentication
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

测试驱动开发优先 - 先编写测试再实现功能；使用pytest-celery验证任务行为
性能优先 - 通过分块处理、连接池、合理预取优化吞吐量
可靠性保障 - 任务重试、确认策略、无任务丢失
可扩展性 - 分布式Worker、任务路由、自动扩缩容、队列优先级
安全第一 - 签名任务、安全序列化、中间件认证
可观测性 - 全面监控、指标追踪、链路追踪、告警机制

风险等级：中等

任务处理失败可能影响业务运营
不当的序列化（如Pickle）可能导致代码执行漏洞
缺少重试/超时机制可能导致任务堆积和系统性能下降
中间件配置错误可能导致任务丢失或消息泄露

2. Implementation Workflow (TDD)

2. 实现工作流（测试驱动开发）

Step 1: Write Failing Test First

步骤1：先编写失败的测试用例

python

undefined

python

undefined

tests/test_tasks.py

import pytest from celery.contrib.testing.tasks import ping from celery.result import EagerResult

@pytest.fixture def 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

undefined

import pytest from celery.contrib.testing.tasks import ping from celery.result import EagerResult

@pytest.fixture def 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): """测试订单处理返回正确结果""" from myapp.tasks import process_order

    # 执行任务
    result = process_order.delay(order_id=123)

    # 断言预期行为
    assert result.get(timeout=10) == {
        'order_id': 123,
        'status': 'success'
    }

def test_process_order_idempotent(self, celery_app, celery_worker):
    """测试任务幂等性 - 可安全重试"""
    from myapp.tasks import process_order

    # 执行两次
    result1 = process_order.delay(order_id=123).get(timeout=10)
    result2 = process_order.delay(order_id=123).get(timeout=10)

    # 重试应安全无副作用
    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):
    """测试任务在临时失败时自动重试"""
    from myapp.tasks import process_order

    # 模拟首次失败，第二次成功
    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

undefined

Step 2: Implement Minimum to Pass

步骤2：实现最小功能以通过测试

python

undefined

python

undefined

myapp/tasks.py

from 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)

undefined

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

    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)

undefined

Step 3: Refactor Following Patterns

步骤3：遵循模式进行重构

Add proper error handling, time limits, and observability.

添加完善的错误处理、时间限制和可观测性配置。

Step 4: Run Full Verification

步骤4：执行完整验证

bash

undefined

bash

undefined

Run all Celery tests

运行所有Celery测试

pytest tests/test_tasks.py -v

Run with coverage

运行测试并生成覆盖率报告

pytest tests/test_tasks.py --cov=myapp.tasks --cov-report=term-missing

Test workflow patterns

测试工作流模式

pytest tests/test_workflows.py -v

Integration test with real broker

使用真实中间件进行集成测试

pytest tests/integration/ --broker=redis://localhost:6379/0

---

pytest tests/integration/ --broker=redis://localhost:6379/0

---

3. Performance Patterns

3. 性能优化模式

Pattern 1: Task Chunking

模式1：任务分块处理

python

undefined

python

undefined

Bad - Individual tasks for each item

不良实践 - 为每个项创建单独任务

for item_id in item_ids: # 10,000 items = 10,000 tasks process_item.delay(item_id)

for item_id in item_ids: # 10,000个项 = 10,000个任务 process_item.delay(item_id)

Good - Process in batches

最佳实践 - 批量处理

@app.task def 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

@app.task def process_batch(item_ids: list): """分块处理项以提升效率""" results = [] for chunk in chunks(item_ids, size=100): items = fetch_items_bulk(chunk) # 单次数据库查询 results.extend([process(item) for item in items]) return results

Dispatch in chunks

分块分发任务

for chunk in chunks(item_ids, size=100): process_batch.delay(chunk) # 100 tasks instead of 10,000

undefined

for chunk in chunks(item_ids, size=100): process_batch.delay(chunk) # 100个任务替代10,000个

undefined

Pattern 2: Prefetch Tuning

模式2：预取参数调优

python

undefined

python

undefined

Bad - Default prefetch for I/O-bound tasks

不良实践 - I/O密集型任务使用默认预取配置

app.conf.worker_prefetch_multiplier = 4 # Too many reserved

app.conf.worker_prefetch_multiplier = 4 # 预留任务过多

Good - Tune based on task type

最佳实践 - 根据任务类型调整

CPU-bound: Higher prefetch, fewer workers

CPU密集型：更高预取数，更少Worker

app.conf.worker_prefetch_multiplier = 4

celery -A app worker --concurrency=4

I/O-bound: Lower prefetch, more workers

I/O密集型：更低预取数，更多Worker

app.conf.worker_prefetch_multiplier = 1

celery -A app worker --pool=gevent --concurrency=100

Long tasks: Disable prefetch

长耗时任务：禁用预取

app.conf.worker_prefetch_multiplier = 1 app.conf.task_acks_late = True

undefined

app.conf.worker_prefetch_multiplier = 1 app.conf.task_acks_late = True

undefined

Pattern 3: Result Backend Optimization

模式3：结果后端优化

python

undefined

python

undefined

Bad - Storing results for fire-and-forget tasks

不良实践 - 为无需结果的任务存储结果

@app.task def send_email(to, subject, body): mailer.send(to, subject, body) return {'sent': True} # Stored in Redis unnecessarily

@app.task def send_email(to, subject, body): mailer.send(to, subject, body) return {'sent': True} # 不必要地存储在Redis中

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 need

最佳实践 - 为需要的结果设置过期时间

app.conf.result_expires = 3600 # 1 hour

app.conf.result_expires = 3600 # 1小时

Good - Store minimal data, reference external storage

最佳实践 - 仅存储最小数据，引用外部存储

@app.task def 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

undefined

@app.task def process_large_file(file_id): data = process(read_file(file_id)) result_key = save_to_s3(data) # 大结果存储在外部 return {'result_key': result_key} # 仅存储引用

undefined

Pattern 4: Connection Pooling

模式4：连接池配置

python

undefined

python

undefined

Bad - Creating new connections per task

不良实践 - 每个任务创建新连接

@app.task def query_database(query): conn = psycopg2.connect(...) # New connection each time result = conn.execute(query) conn.close() return result

@app.task def query_database(query): conn = psycopg2.connect(...) # 每次任务创建新连接 result = conn.execute(query) conn.close() return result

Good - Use connection pools

最佳实践 - 使用连接池

from sqlalchemy import create_engine from redis import ConnectionPool, Redis

Initialize once at module level

在模块级别初始化一次

db_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.task def query_database(query): with db_engine.connect() as conn: # Uses pool return conn.execute(query).fetchall()

@app.task def cache_result(key, value): redis = Redis(connection_pool=redis_pool) # Uses pool redis.set(key, value)

undefined

db_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.task def query_database(query): with db_engine.connect() as conn: # 使用连接池 return conn.execute(query).fetchall()

@app.task def cache_result(key, value): redis = Redis(connection_pool=redis_pool) # 使用连接池 redis.set(key, value)

undefined

Pattern 5: Task Routing

模式5：任务路由配置

python

undefined

python

undefined

Bad - All tasks in single queue

不良实践 - 所有任务放入单个队列

@app.task def critical_payment(): pass

@app.task def generate_report(): pass # Blocks payment processing

@app.task def critical_payment(): pass

@app.task def generate_report(): pass # 阻塞支付任务处理

Good - Route to dedicated queues

最佳实践 - 路由到专用队列

from 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'}, }

from kombu import Queue, Exchange

app.conf.task_routes = { 'tasks.critical_payment': {'queue': 'critical'}, 'tasks.generate_report': {'queue': 'bulk'}, }

Run dedicated workers per queue

为每个队列启动专用Worker

celery -A app worker -Q critical --concurrency=4

celery -A app worker -Q bulk --concurrency=2

---

---

4. Core Responsibilities

4. 核心职责

1. Task Design & Workflow Orchestration

1. 任务设计与工作流编排

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)

使用合适的装饰器定义任务（
```
@app.task
```
、
```
@shared_task
```
）
实现幂等性任务（可安全重试）
使用Chains实现顺序执行、Groups实现并行执行、Chords实现Map-Reduce
设计任务路由到特定队列/Worker
避免长耗时任务（拆分为子任务）

2. Broker Configuration & Management

2. 中间件配置与管理

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

简单场景选Redis，高可靠场景选RabbitMQ
配置连接池、心跳机制和故障转移
启用中间件认证与加密（TLS）
监控中间件健康状态与连接状态

3. Task Reliability & Error Handling

3. 任务可靠性与错误处理

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

实现带指数退避的重试逻辑
关键任务使用
```
acks_late=True
```
设置合理的任务时间限制（软/硬超时）
通过错误回调优雅处理异常
为失败任务配置死信队列
设计幂等性任务以安全处理重试

4. Result Backends & State Management

4. 结果后端与状态管理

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)

选择合适的结果后端（Redis、数据库、RPC）
设置结果过期时间以防止内存泄漏
无需结果的任务使用
```
ignore_result=True
```
仅在结果中存储最小数据（使用外部存储）

5. Celery Beat Scheduling

5. Celery Beat调度配置

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

为周期性任务定义Crontab调度
简单周期任务使用间隔调度
配置Beat调度器持久化（数据库后端）
使用任务锁避免调度冲突

6. Monitoring & Observability

6. 监控与可观测性

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

部署Flower进行实时监控
导出Prometheus指标用于告警
跟踪任务成功率/失败率与队列长度
实现分布式链路追踪（关联ID）
带上下文信息记录任务执行日志

5. Implementation Patterns

5. 实现模式

Pattern 1: Task Definition Best Practices

模式1：任务定义最佳实践

python

undefined

python

undefined

COMPLETE TASK DEFINITION

完整任务定义示例

from celery import Celery from celery.exceptions import SoftTimeLimitExceeded import 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

undefined

from celery import Celery from celery.exceptions import SoftTimeLimitExceeded import 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): """带完善错误处理与重试机制的订单处理任务""" 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

undefined

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

模式2：工作流模式（Chains、Groups、Chords）

python

from celery import chain, group, chord

python

from celery import chain, group, chord

CHAIN: Sequential execution (A -> B -> C)

CHAIN：顺序执行（A -> B -> C）

workflow = chain( fetch_data.s('https://api.example.com/data'), process_item.s(), send_notification.s() )

GROUP: Parallel execution

GROUP：并行执行

job = group(fetch_data.s(url) for url in urls)

CHORD: Map-Reduce (parallel + callback)

CHORD：Map-Reduce（并行执行 + 回调）

workflow = chord( group(process_item.s(item) for item in items) )(aggregate_results.s())

undefined

workflow = chord( group(process_item.s(item) for item in items) )(aggregate_results.s())

undefined

Pattern 3: Production Configuration

模式3：生产环境配置

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,
)

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

模式4：重试策略与错误处理

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()

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):
    """遇到RequestException时自动重试，带指数退避"""
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

Pattern 5: Celery Beat Scheduling

模式5：Celery Beat调度配置

python

from celery.schedules import crontab
from 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),
    },
}

python

from celery.schedules import crontab
from 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. 安全标准

6.1 Secure Serialization

6.1 安全序列化

python

undefined

python

undefined

DANGEROUS: Pickle allows code execution

危险：Pickle可能导致代码执行漏洞

app.conf.task_serializer = 'pickle' # NEVER!

app.conf.task_serializer = 'pickle' # 绝对禁止！

SECURE: Use JSON

安全：使用JSON

app.conf.update( task_serializer='json', result_serializer='json', accept_content=['json'], )

undefined

app.conf.update( task_serializer='json', result_serializer='json', accept_content=['json'], )

undefined

6.2 Broker Authentication & TLS

6.2 中间件认证与TLS

python

undefined

python

undefined

Redis with TLS

带TLS的Redis配置

app.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 TLS

带TLS的RabbitMQ配置

app.conf.broker_url = 'amqps://user:password@localhost:5671/vhost'

undefined

app.conf.broker_url = 'amqps://user:password@localhost:5671/vhost'

undefined

6.3 Input Validation

6.3 输入验证

python

from pydantic import BaseModel

class OrderData(BaseModel):
    order_id: int
    amount: float

@app.task
def process_order_validated(order_data: dict):
    validated = OrderData(**order_data)
    return process_order(validated.dict())

python

from pydantic import BaseModel

class OrderData(BaseModel):
    order_id: int
    amount: float

@app.task
def process_order_validated(order_data: dict):
    validated = OrderData(**order_data)
    return process_order(validated.dict())

7. Common Mistakes

7. 常见错误

Mistake 1: Using Pickle Serialization

错误1：使用Pickle序列化

python

undefined

python

undefined

DON'T

禁止

app.conf.task_serializer = 'pickle'

DO

DON'T: Retries increment multiple times

禁止：重试会导致计数器多次递增

@app.task def increment_counter(user_id): user.counter += 1 user.save()

DO: Safe to retry

推荐：重试安全无副作用

@app.task def set_counter(user_id, value): user.counter = value user.save()

undefined

@app_task def set_counter(user_id, value): user.counter = value user.save()

undefined

Mistake 3: Missing Time Limits

错误3：未设置时间限制

python

undefined

python

undefined

DON'T

禁止

@app.task def slow_task(): external_api_call()

DO

DON'T

禁止

@app.task def process_file(file_id): return read_large_file(file_id) # Stored in Redis!

@app.task def process_file(file_id): return read_large_file(file_id) # 结果会存储在Redis中！