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中使用多線程的。