Python使用celery

Python3.7
celery5.0.5

celery

Celery 是一個簡單，靈活，可靠的分布式系統，用于處理大量訊息，同時為操作提供維護此類系統所需的工具，它是一個任務佇列，專注于實時處理，同時還支持任務調度，
安裝模塊

pip install celery==5.0.5
pip install redis=3.5.3  # redis作為backend，需安裝驅動模塊

異步任務

import time

from celery import Celery

broker = 'pyamqp://admin:admin@localhost//'  # rabbitmq作為broker
# broker = 'redis://localhost:6379/1'  # redis作為broker
backend = 'redis://localhost:6379/2'
app = Celery('tasks', broker=broker, backend=backend)


@app.task
def add(x, y):
    time.sleep(10)
    res = x + y
    print('add:{}'.format(res))
    return res


if __name__ == '__main__':
    r = add.delay(5, 8)
    print('r:', r)
    print('ready:', r.ready())
    print('result:', r.result)
    print('get:', r.get())

啟動celery服務

$ celery -A tasks worker --loglevel=INFO
 
 -------------- celery@MacMiniMD389deMac-mini.local v5.0.5 (singularity)
--- ***** ----- 
-- ******* ---- Darwin-18.7.0-x86_64-i386-64bit 2020-12-25 15:27:36
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x107f3c550
- ** ---------- .> transport:   amqp://admin:**@localhost:5672//
- ** ---------- .> results:     redis://localhost:6379/2
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                

[tasks]
  . tasks.add

[2020-12-25 15:27:37,138: INFO/MainProcess] Connected to amqp://admin:**@127.0.0.1:5672//
[2020-12-25 15:27:37,155: INFO/MainProcess] mingle: searching for neighbors
[2020-12-25 15:27:38,210: INFO/MainProcess] mingle: all alone
[2020-12-25 15:27:38,251: INFO/MainProcess] celery@MacMiniMD389deMac-mini.local ready.

后臺啟動

# 啟動
$ sudo celery multi start -A tasks worker --loglevel=INFO
Password:
celery multi v5.0.5 (singularity)
> Starting nodes...
        > worker@MacMiniMD389deMac-mini.local: OK
$ 
$ # 重啟
$ sudo celery multi restart -A tasks worker --loglevel=INFO
celery multi v5.0.5 (singularity)
> Stopping nodes...
        > worker@MacMiniMD389deMac-mini.local: TERM -> 15391
> Waiting for 1 node -> 15391.....
        > worker@MacMiniMD389deMac-mini.local: OK
> Restarting node worker@MacMiniMD389deMac-mini.local: OK
> Waiting for 1 node -> None...
$
$ # 停止(異步)
$ sudo celery multi stop -A tasks worker --loglevel=INFO
celery multi v5.0.5 (singularity)
> Stopping nodes...
        > worker@MacMiniMD389deMac-mini.local: TERM -> 15583
(celery5_env) MacMiniMD389deMac-mini at ~/workspace
$
$ # 停止（同步）
(celery5_env) MacMiniMD389deMac-mini at ~/workspace/celery_project ? sudo celery multi stopwait -A tasks worker --loglevel=INFO 
celery multi v5.0.5 (singularity)
> Stopping nodes...
        > worker@MacMiniMD389deMac-mini.local: TERM -> 16919
> Waiting for 1 node -> 16919.....
        > worker@MacMiniMD389deMac-mini.local: OK
> worker@MacMiniMD389deMac-mini.local: DOWN
> Waiting for 1 node -> None...
$

默認情況下，celery multi start啟動命令會在當前目錄中創建pid和日志檔案，為了防止多個worker相互重疊啟動，設定這些worker的啟動檔案放在專用目錄中：

$ sudo celery multi start -A tasks worker --loglevel=INFO --pidfile=/var/run/celery/%n.pid --logfile=/var/log/celery/%n%I.log
Password:
celery multi v5.0.5 (singularity)
> Starting nodes...
        > worker@MacMiniMD389deMac-mini.local: OK
(celery5_env) MacMiniMD389deMac-mini at ~/workspace/celery_project ? 
$
$ ls /var/run/celery
worker.pid
$ ls /var/log/celery 
worker-1.log worker-2.log worker-3.log worker-4.log worker-5.log worker-6.log worker-7.log worker-8.log worker.log
$

查看celery multi 官方示例

執行異步任務

>>> from tasks import add
>>> 
>>> res = add.delay(5, 8)
>>> 
>>> res
<AsyncResult: b9f9cbc8-ffb2-4cb7-827f-2f0e724d20f3>
>>> 
>>> res.ready()  # 任務是否完成
False
>>> res.ready()
False
>>> res.result  # 任務回傳值
>>> 
>>> res.result
13
>>> # result.get()  # 阻塞等待回傳結果
>>>> res.get(timeout=3)  # 等待3s，3s后任務未完成則拋出timeout例外
Traceback (most recent call last):
  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/celery/backends/asynchronous.py", line 267, in _wait_for_pending
    on_interval=on_interval):
  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/celery/backends/asynchronous.py", line 52, in drain_events_until
    raise socket.timeout()
socket.timeout

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/celery/result.py", line 230, in get
    on_message=on_message,
  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/celery/backends/asynchronous.py", line 199, in wait_for_pending
    for _ in self._wait_for_pending(result, **kwargs):
  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/celery/backends/asynchronous.py", line 271, in _wait_for_pending
    raise TimeoutError('The operation timed out.')
celery.exceptions.TimeoutError: The operation timed out.
>>> res.get(timeout=3)
13
>>>
>>>
>>>
>>>
>>># 例外
>>>> res = add.delay(5, '8')
>>> res.get()  # 如果任務引發例外，get（）將重新引發例外
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/celery/result.py", line 219, in get
    self.maybe_throw(callback=callback)
  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/celery/result.py", line 335, in maybe_throw
    self.throw(value, self._to_remote_traceback(tb))
  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/celery/result.py", line 328, in throw
    self.on_ready.throw(*args, **kwargs)
  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/vine/promises.py", line 234, in throw
    reraise(type(exc), exc, tb)
  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/vine/utils.py", line 29, in reraise
    raise value.with_traceback(tb)
TypeError: unsupported operand type(s) for +: 'int' and 'str'
>>> 
>>> res.get(propagate=False)  # 可以通過引數propagate是否重新引發例外
TypeError("unsupported operand type(s) for +: 'int' and 'str'")
>>> 
>>> res.traceback  # 查看例外資訊
'Traceback (most recent call last):\n  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/celery/app/trace.py", line 405, in trace_task\n    R = retval = fun(*args, **kwargs)\n  File "/Users/macminimd389/.virtualenvs/celery5_env/lib/python3.7/site-packages/celery/app/trace.py", line 697, in __protected_call__\n    return self.run(*args, **kwargs)\n  File "/Users/macminimd389/workspace/celery_project/tasks.py", line 19, in add\n    res = x + y\nTypeError: unsupported operand type(s) for +: \'int\' and \'str\'\n'
>>> 
>>>