Celery 4.3.0 在task中执行多线程任务

测试Celery任务可否使用多线程

在开发的调试过程当中，发现若是在django项目里面或者celery的task中使用协程gevent的话，使用monkey补丁的时候会报错。

那么尝试了好久，发如今celery中是能够执行多线程的，下面来演示一下执行的示例。

编写使用多线程的task

import threading
from time import sleep,ctime

def smoke():
    for i in range(3):
        print("smoke...%d"%i)
        sleep(10)

def drunk():
    for i in range(3):
        print("drink...%d"%i)
        sleep(10)


@celery_app.task
def test_use_thread():
    print('---开始---:%s' % ctime())

    t1 = threading.Thread(target=smoke)
    t2 = threading.Thread(target=drunk)

    t1.start()
    t2.start()

    while True:
        length = len(threading.enumerate())
        print("当前运行的线程数为:%d" % length)
        if length <= 1:
            break

        sleep(0.5)

    # sleep(5) #
    print('---结束---:%s' % ctime())
复制代码

启动celery服务

celery -A celery_tasks worker -l info -P eventlet

在django项目的交互模式下执行多线程的task任务

In [1]: from celery_tasks.tasks import test_use_thread

In [2]: test_use_thread.delay()
Out[2]: <AsyncResult: 69bba7b0-3808-47f1-ad36-2fea75930901>
复制代码

查看celery中的执行打印状况

能够看到任务顺利并发执行，也就是Celery虽然自身已是异步任务，仍是能够继续在task中使用多线程的。