celery 常用配置

设置时区

1
CELERY_TIMEZONE = 'Asia/Shanghai'

启动时区设置

1
CELERY_ENABLE_UTC = True

限制任务的执行频率
下面这个就是限制tasks模块下的add函数,每秒钟只能执行10次

1
CELERY_ANNOTATIONS = {'tasks.add':{'rate_limit':'10/s'}}

或者限制所有的任务的刷新频率

1
CELERY_ANNOTATIONS = {'*':{'rate_limit':'10/s'}}

也可以设置如果任务执行失败后调用的函数

1
2
def my_on_failure(self,exc,task_id,args,kwargs,einfo):
print('task failed')
1
CELERY_ANNOTATIONS = {'*':{'on_failure':my_on_failure}}

并发的worker数量,也是命令行-c指定的数目
事实上并不是worker数量越多越好,保证任务不堆积,加上一些新增任务的预留就可以了

1
CELERYD_CONCURRENCY = 20

celery worker每次去redis取任务的数量,默认值就是4

1
CELERYD_PREFETCH_MULTIPLIER = 4

每个worker执行了多少次任务后就会死掉,建议数量大一些

1
CELERYD_MAX_TASKS_PER_CHILD = 200

使用redis作为任务队列

1
2
组成: db+scheme://user:password@host:port/dbname
BROKER_URL = 'redis://127.0.0.1:6379/0'

celery任务执行结果的超时时间

1
CELERY_TASK_RESULT_EXPIRES = 1200

单个任务的运行时间限制,否则会被杀死

1
CELERYD_TASK_TIME_LIMIT = 60

使用redis存储任务执行结果,默认不使用

1
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'

将任务结果使用’pickle’序列化成’json’格式
任务序列化方式

1
CELERY_TASK_SERIALIZER = 'pickle'

任务执行结果序列化方式

1
CELERY_RESULT_SERIALIZER = 'json'

也可以直接在Celery对象中设置序列化方式

1
app = Celery('tasks', broker='...', task_serializer='yaml')

关闭限速

1
CELERY_DISABLE_RATE_LIMITS = True

一份比较常用的配置文件:
在celery4.x以后,就是BROKER_URL,如果是以前,需要写成

1
2
CELERY_BROKER_URL
BROKER_URL = 'redis://127.0.0.1:6379/0'

指定结果的接收地址

1
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'

指定任务序列化方式

1
CELERY_TASK_SERIALIZER = 'msgpack'

指定结果序列化方式

1
CELERY_RESULT_SERIALIZER = 'msgpack'

指定任务接受的序列化类型.

1
CELERY_ACCEPT_CONTENT = ['msgpack']

任务过期时间,celery任务执行结果的超时时间

1
CELERY_TASK_RESULT_EXPIRES = 24 * 60 * 60

任务发送完成是否需要确认,对性能会稍有影响

1
CELERY_ACKS_LATE = True

压缩方案选择,可以是zlib, bzip2,默认是发送没有压缩的数据

1
CELERY_MESSAGE_COMPRESSION = 'zlib'

规定完成任务的时间
在5s内完成任务,否则执行该任务的worker将被杀死,任务移交给父进程

1
CELERYD_TASK_TIME_LIMIT = 5

celery worker的并发数,默认是服务器的内核数目,也是命令行-c参数指定的数目

1
CELERYD_CONCURRENCY = 4

celery worker 每次去BROKER中预取任务的数量

1
CELERYD_PREFETCH_MULTIPLIER = 4

每个worker执行了多少任务就会死掉,默认是无限的

1
CELERYD_MAX_TASKS_PER_CHILD = 40

设置默认的队列名称,如果一个消息不符合其他的队列就会放在默认队列里面,如果什么都不设置的话,数据都会发送到默认的队列中

1
CELERY_DEFAULT_QUEUE = "default"

队列的详细设置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
CELERY_QUEUES = {
"default": { # 这是上面指定的默认队列
"exchange": "default",
"exchange_type": "direct",
"routing_key": "default"
},
"topicqueue": { # 这是一个topic队列 凡是topictest开头的routing key都会被放到这个队列
"routing_key": "topic.#",
"exchange": "topic_exchange",
"exchange_type": "topic",
},
"task_eeg": { # 设置扇形交换机
"exchange": "tasks",
"exchange_type": "fanout",
"binding_key": "tasks",
},

或者配置成下面两种方式:

配置队列(settings.py)

1
2
3
4
5
6
7
8
9
10
11
CELERY_QUEUES = (
Queue('default',
Exchange('default'),
routing_key='default'),
Queue('for_task_collect',
Exchange('for_task_collect'),
routing_key='for_task_collect'),
Queue('for_task_compute',
Exchange('for_task_compute'),
routing_key='for_task_compute'),
)

路由(哪个任务放入哪个队列)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
CELERY_ROUTES = {
'umonitor.tasks.multiple_thread_metric_collector':
{
'queue': 'for_task_collect',
'routing_key': 'for_task_collect'
},
'compute.tasks.multiple_thread_metric_aggregate':
{
'queue': 'for_task_compute',
'routing_key': 'for_task_compute'
},
'compute.tasks.test':
{
'queue': 'for_task_compute',
'routing_key': 'for_task_compute'
},
}