生產者:

• 文件1: 定義任務

#!/usr/bin/env python3
# coding: utf-8
#
# Created by dylanchu on 19-1-6

from celery import Celery
import settings

pw = settings.SESSION_REDIS[‘password‘]
celery_broker = ‘redis://:%s@localhost:6379/0‘ % pw
celery_backend = celery_broker

app = Celery(‘tasks‘, broker=celery_broker, backend=celery_backend)

@app.task
def analysis_main_12(current_id_str, q_num_str):
    pass

@app.task
def analysis_main_3(current_id_str, q_num_str):
    pass
 

• 文件2: 產生任務並放到隊列

#!/usr/bin/env python3
# coding: utf-8

from celery_tasks import analysis_main_12, analysis_main_3

def main():
    q = get_q_from_db()
    try:
        if q.q_type == 3 or q.q_type == ‘3‘:
            ret = analysis_main_3.apply_async(args=(str(current_test.id), str(q_num)), queue=‘for_q_type3‘)
        else:
            ret = analysis_main_12.apply_async(args=(str(current_test.id), str(q_num)), queue=‘for_q_type12‘)
        logging.info("AsyncResult id: %s" % ret.id)
    except Exception as e:
        logging.error(‘upload_success: celery tasks enqueue: ERROR: shell ret: %s‘ % e)

if __name__ == ‘__main__‘:
    main()
 

消費者(worker)

celery使用多隊列