---

---

【Python３】Celeryによるタスクのスケジューリング - Part2

投稿日：2020年10月30日

Celeryを使用すると柔軟なタスクのスケジューリングをすることができます。この記事ではCeleryのための環境構築と簡単な使い方について解説していきます。

---

はじめに

ソースコード

git clone https://github.com/ogipochi/python_celery_example.git -b part2

---

環境構築

設定の定義

python_app/src/
└── proj
    ├── __init__.py
    ├── celery.py
    └── tasks.py

from celery import Celery

# Celeryオブジェクトを作成
app = Celery('proj',
             broker='redis://redis_app:6379/',
             backend='redis://redis_app:6379/',
             include=['proj.tasks',])

# タスクの結果の保持時間を変更
app.conf.update(
    result_expires=3600
)

if __name__ == '__main__':
    app.start()

from .celery import app

@app.task
def add(x, y):
    return x + y

@app.task
def mul(x, y):
    return x * y

@app.task
def xsum(numbers):
    return sum(numbers)

FROM python:3.7

RUN pip install -U celery[redis]

COPY ./src /src

WORKDIR /src

ENTRYPOINT [ "celery","-A","proj","worker","--loglevel=INFO" ]

起動

$ sudo docker-compose up --build

[...]

python_app_1  | /usr/local/lib/python3.7/site-packages/celery/platforms.py:798: RuntimeWarning: You're running the worker with superuser privileges: this is
python_app_1  | absolutely not recommended!
python_app_1  | 
python_app_1  | Please specify a different user using the --uid option.
python_app_1  | 
python_app_1  | User information: uid=0 euid=0 gid=0 egid=0
python_app_1  | 
python_app_1  |   uid=uid, euid=euid, gid=gid, egid=egid,
python_app_1  |  
python_app_1  |  -------------- celery@0b5c0ff894a3 v5.0.1 (singularity)
python_app_1  | --- ***** ----- 
python_app_1  | -- ******* ---- Linux-5.4.0-51-generic-x86_64-with-debian-9.8 2020-10-29 12:16:53
python_app_1  | - *** --- * --- 
python_app_1  | - ** ---------- [config]
python_app_1  | - ** ---------- .> app:         proj:0x7fbf282c6390
python_app_1  | - ** ---------- .> transport:   redis://redis_app:6379//
python_app_1  | - ** ---------- .> results:     redis://redis_app:6379/
python_app_1  | - *** --- * --- .> concurrency: 4 (prefork)
python_app_1  | -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
python_app_1  | --- ***** ----- 
python_app_1  |  -------------- [queues]
python_app_1  |                 .> celery           exchange=celery(direct) key=celery
python_app_1  |                 
python_app_1  | 
python_app_1  | [tasks]
python_app_1  |   . proj.tasks.add
python_app_1  |   . proj.tasks.mul
python_app_1  |   . proj.tasks.xsum
python_app_1  | 
python_app_1  | [2020-10-29 12:16:53,592: INFO/MainProcess] Connected to redis://redis_app:6379//
python_app_1  | [2020-10-29 12:16:53,600: INFO/MainProcess] mingle: searching for neighbors
python_app_1  | [2020-10-29 12:16:54,618: INFO/MainProcess] mingle: all alone
python_app_1  | [2020-10-29 12:16:54,630: INFO/MainProcess] celery@0b5c0ff894a3 ready.

---

実践

タスクの実行

$ sudo docker exec -it celery_sample_python_app_1 bash
root@0b5c0ff894a3:/src# python
Python 3.7.3 (default, Mar 27 2019, 23:40:30) 
[GCC 6.3.0 20170516] on linux
Type "help", "copyright", "credits" or "license" for more information.

>>> from proj.tasks import add
>>> add.delay(5,5)
<AsyncResult: e52a5691-9249-4ea4-a0ae-444fd3b742e6>

>>> add.apply_async((5,5))
<AsyncResult: 6bfa49e7-f89b-40d2-9ec7-2840b2dea8a7>

>>> result = add.apply_async((4,3), countdown=10)
>>> result.ready()
False
>>> result.ready()
True
>>> result.get(timeout=1)
7

>>> from datetime import datetime, timedelta
>>> result = add.apply_async((3,3), eta=datetime.now() + timedelta(seconds = 10))
>>> result.ready()
False
>>> result.ready()
True
>>> result.get(timeout=1)
6

>>> add(4,5)
9

>>> result = add.apply_async(({"a": 1, "b" : 2},3), eta=datetime.now() + timedelta(seconds = 10))
>>> result.failed()
True
>>> result.successful()
False
>>> result.state
'FAILURE'

>>> my_task = add.signature((2,2), countdown=10)
>>> my_task.delay()
<AsyncResult: 6c317308-9290-4f5e-acfd-b68bf962e570>

---

参考書籍

---

---

---