一份比较常用的配置文件:

# 在 celery4.x 以后,就是 BROKER_URL,如果是以前,需要写成 CELERY_BROKER_URL
BROKER_URL = 'redis://127.0.0.1:6379/0'

# 指定结果的接收地址
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'

# 指定任务序列化方式
CELERY_TASK_SERIALIZER = 'msgpack'

# 指定结果序列化方式
CELERY_RESULT_SERIALIZER = 'msgpack'

# 指定任务接受的序列化类型.
CELERY_ACCEPT_CONTENT = ['msgpack']

# 任务过期时间,celery 任务执行结果的超时时间
CELERY_TASK_RESULT_EXPIRES = 24 * 60 * 60

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

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

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

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

# celery worker 每次去 BROKER 中预取任务的数量
CELERYD_PREFETCH_MULTIPLIER = 4

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

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

# 队列的详细设置

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",
    },

}
# 当使用路由时,手动指定queue即可


或者配置成:

# 配置队列(settings.py)

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

# 自动路由(哪个任务放入哪个队列)
CELERY_ROUTES = {
    '函数名':
    {
        'queue': 'for_task_collect',
        'routing_key': 'for_task_collect'
    },
    '函数名':
    {
        'queue': 'for_task_compute',
        'routing_key': 'for_task_compute'
    },
    '函数名':
    {
         'queue': 'for_task_compute',
         'routing_key': 'for_task_compute'
    },
}
# 可在开启时查看到对应的函数task和queue
## 需要在 对应的函数前的
# @clus.task(name="regex") 中指定名字
# 否则就是 cluster.task.函数名 对应也得修改