さくらのクラウド シンプルMQを使ってDjangoのCeleryを動かす

「シンプルMQ」が2025年2月6日にさくらのクラウドに追加されました。

> 「シンプルMQ」はソフトウェア同士を非同期に連携するために、ソフトウェアコンポーネント間でのデータの送受信ができるマネージド型のメッセージキューサービスです。

Pythonを使って非同期の処理を複数のワーカーで分散処理する場合には、タスクキューとして [Celery](https://docs.celeryq.dev/en/stable/index.html) を使うことがよくあると思います。

Celeryでは実際に利用するキューとしてオンメモリやデータベース、Amazon SQS、Redisなどのキューを利用することができます。

同様にしてさくらのクラウドのシンプルMQをCeleryのバックエンドとして使えるように、ドライバ [chibiegg/kombu-sakura-simplemq](https://github.com/chibiegg/kombu-sakura-simplemq) を書きましたので紹介します。
(正確にはCeleryが利用しているMQのラッパーである [kombu](https://github.com/celery/kombu) のTransportです)

Celeryは利用したことがある方をメインターゲットに書いています。Celery自体の使い方については他の情報もご参照ください。


# インストール

[PyPI](https://pypi.org/project/kombu-sakura-simplemq/) にパッケージを登録していますので、pipでインストールできます。

```bash
pip install kombu-sakura-simplemq
```

# さくらのクラウドの準備

事前にさくらのクラウドのコントロールパネルでシンプルMQのキューを作成しておきます。


キュー名はこの後使いますので、控えておいてください。
何のキューなのかわかりやすい名前をつけておきます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/73788/1cbd9474-515e-4eab-a94e-b1d60ef1b2c9.png)


キューを作成するとメッセージの送受信に必要なAPIキーが表示されます。
これも利用しますので控えておいてください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/73788/2419a0fb-0ba4-4896-a578-4dc20cf4c769.png)


# DjangoにCeleryを設定する

## Celeryパッケージのインストール

```bash
pip install celery
```

## settings.pyにCeleryの設定を追加

```python:settings.py
from kombu import transport
transport.TRANSPORT_ALIASES["sakura-simplemq"] = "kombu_sakura_simplemq.transport:Transport"

# celery configurations
CELERY_BROKER_URL = "sakura-simplemq://:{}@".format(
    os.environ.get("SIMPLE_MQ_API_KEY", ""),
)
CELERY_TASK_DEFAULT_QUEUE = "test-queue"  # 作成しておいたキューの名前
CELERY_RESULT_BACKEND = None
CELERY_ACCEPT_CONTENT = ["application/json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_TIMEZONE = "Asia/Tokyo"

```

## celery.pyを追加

```python:djangoproject/celery.py
import os
import django
from django.conf import settings

# celeryで使うDjangoの設定ファイル(settings.py)を指定
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "djangoproject.settings")
django.setup()

from celery import Celery  # noqa: E402

app = Celery("djangoproject")

# Djangoのconfigファイルをceleryのconfigとして使う宣言
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

# 利用されるタスクをimportしておく
from djangoproject.tasks import test_task  # noqa: E402,F401
```

## タスクを定義

```python:djangoproject/tasks.py
import logging

from djangoproject.celery import app

logger = logging.getLogger(__name__)


@app.task
def test_task():
    logger.debug("テストタスクが実行されました")
```

# ジョブの実行

非同期に実行するには定義したタスクを `delay()` で呼び出します。
非同期でタスクを実行すると、シンプルMQのキューにメッセージが蓄積されます。

```python
from djangoproject.tasks import test_task

test_task.delay()
```

# ワーカーの起動

他のキューを利用した場合と同様にCeleryのワーカーを起動すると、先ほど蓄積したタスクが実行されます。

```bash
celery -A djangoproject worker -l DEBUG
```


```text:実行例
% celery -A djangoproject worker -l DEBUG
[2025-02-23 11:56:06,765: DEBUG/MainProcess] | Worker: Preparing bootsteps.
[2025-02-23 11:56:06,782: DEBUG/MainProcess] | Worker: Building graph...
[2025-02-23 11:56:06,783: DEBUG/MainProcess] | Worker: New boot order: {Beat, StateDB, Timer, Hub, Pool, Autoscaler, Consumer}
[2025-02-23 11:56:06,848: DEBUG/MainProcess] | Consumer: Preparing bootsteps.
[2025-02-23 11:56:06,849: DEBUG/MainProcess] | Consumer: Building graph...
[2025-02-23 11:56:06,883: DEBUG/MainProcess] | Consumer: New boot order: {Connection, Events, Heart, Agent, Mingle, Gossip, Tasks, Control, event loop}
[2025-02-23 11:56:06,884: WARNING/MainProcess] No hostname was supplied. Reverting to default 'localhost'
 
 -------------- celery@PCNAME v5.4.0 (opalescent)
--- ***** ----- 
-- ******* ---- macOS-15.3-arm64-arm-64bit 2025-02-23 11:56:06
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         djangoproject:0x108fe1310
- ** ---------- .> transport:   sakura-simplemq://:**@localhost//
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 12 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> test-queue       exchange=test-queue(direct) key=test-queue
                

[tasks]
  . celery.accumulate
  . celery.backend_cleanup
  . celery.chain
  . celery.chord
  . celery.chord_unlock
  . celery.chunks
  . celery.group
  . celery.map
  . celery.starmap
  . djangoproject.tasks.test_task

[2025-02-23 11:56:06,935: DEBUG/MainProcess] | Worker: Starting Pool
[2025-02-23 11:56:07,010: DEBUG/MainProcess] ^-- substep ok
[2025-02-23 11:56:07,011: DEBUG/MainProcess] | Worker: Starting Consumer
[2025-02-23 11:56:07,013: DEBUG/MainProcess] | Consumer: Starting Connection
whether broker connection retries are made during startup in Celery 6.0 and above.
If you wish to retain the existing behavior for retrying connections on startup,
you should set broker_connection_retry_on_startup to True.
  warnings.warn(

[2025-02-23 11:56:07,032: WARNING/MainProcess] No hostname was supplied. Reverting to default 'localhost'
[2025-02-23 11:56:07,032: INFO/MainProcess] Connected to sakura-simplemq://:**@localhost//
[2025-02-23 11:56:07,032: DEBUG/MainProcess] ^-- substep ok
[2025-02-23 11:56:07,032: DEBUG/MainProcess] | Consumer: Starting Events
whether broker connection retries are made during startup in Celery 6.0 and above.
If you wish to retain the existing behavior for retrying connections on startup,
you should set broker_connection_retry_on_startup to True.
  warnings.warn(

[2025-02-23 11:56:07,033: DEBUG/MainProcess] ^-- substep ok
[2025-02-23 11:56:07,033: DEBUG/MainProcess] | Consumer: Starting Heart
[2025-02-23 11:56:07,034: DEBUG/MainProcess] ^-- substep ok
[2025-02-23 11:56:07,034: DEBUG/MainProcess] | Consumer: Starting Tasks
[2025-02-23 11:56:07,034: DEBUG/MainProcess] Timer wake-up! Next ETA 1.0 secs.
[2025-02-23 11:56:07,045: DEBUG/MainProcess] ^-- substep ok
[2025-02-23 11:56:07,045: DEBUG/MainProcess] | Consumer: Starting event loop
[2025-02-23 11:56:07,045: INFO/MainProcess] celery@PCNAME ready.
[2025-02-23 11:56:07,045: DEBUG/MainProcess] basic.qos: prefetch_count->48
[2025-02-23 11:56:07,048: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:07,170: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "GET /v1/queues/test-queue/messages HTTP/1.1" 200 1507
[2025-02-23 11:56:07,175: INFO/MainProcess] Task djangoproject.tasks.test_task[51a56eef-ce55-467a-8950-38df307895b0] received
[2025-02-23 11:56:07,175: DEBUG/MainProcess] TaskPool: Apply <function fast_trace_task at 0x103e8aac0> (args:('djangoproject.tasks.test_task', '51a56eef-ce55-467a-8950-38df307895b0', {'lang': 'py', 'task': 'djangoproject.tasks.test_task', 'id': '51a56eef-ce55-467a-8950-38df307895b0', 'shadow': None, 'eta': None, 'expires': None, 'group': None, 'group_index': None, 'retries': 0, 'timelimit': [None, None], 'root_id': '51a56eef-ce55-467a-8950-38df307895b0', 'parent_id': None, 'argsrepr': '()', 'kwargsrepr': '{}', 'origin': 'gen98617@PCNAME', 'ignore_result': False, 'replaced_task_nesting': 0, 'stamped_headers': None, 'stamps': {}, 'properties': {'correlation_id': '51a56eef-ce55-467a-8950-38df307895b0', 'reply_to': '8df5f54e-6ddb-32b3-a3e1-2fc2991d457e', 'delivery_mode': 2, 'delivery_info': {'exchange': '', 'routing_key': 'test-queue'}, 'priority': 0, 'body_encoding': 'base64', 'delivery_tag': '019530b9-ddfd-7520-93f0-8d22b8b0e616'}, 'reply_to': '8df5f54e-6ddb-32b3-a3e1-2fc2991d457e', 'correlation_id': '51a56eef-ce55-467a-8950-38df307895b0', 'hostname': 'celery@PCNAME', 'delivery_info': {'exchange': '', 'routing_key':... kwargs:{})
[2025-02-23 11:56:07,177: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:07,178: DEBUG/ForkPoolWorker-1] テストタスクが実行されました
[2025-02-23 11:56:07,180: INFO/ForkPoolWorker-1] Task djangoproject.tasks.test_task[51a56eef-ce55-467a-8950-38df307895b0] succeeded in 0.0026704999909270555s: None
[2025-02-23 11:56:07,267: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "GET /v1/queues/test-queue/messages HTTP/1.1" 200 1507
[2025-02-23 11:56:07,268: INFO/MainProcess] Task djangoproject.tasks.test_task[642cb8dc-da32-4061-bbe8-2870f7bdd5c9] received
[2025-02-23 11:56:07,269: DEBUG/MainProcess] TaskPool: Apply <function fast_trace_task at 0x103e8aac0> (args:('djangoproject.tasks.test_task', '642cb8dc-da32-4061-bbe8-2870f7bdd5c9', {'lang': 'py', 'task': 'djangoproject.tasks.test_task', 'id': '642cb8dc-da32-4061-bbe8-2870f7bdd5c9', 'shadow': None, 'eta': None, 'expires': None, 'group': None, 'group_index': None, 'retries': 0, 'timelimit': [None, None], 'root_id': '642cb8dc-da32-4061-bbe8-2870f7bdd5c9', 'parent_id': None, 'argsrepr': '()', 'kwargsrepr': '{}', 'origin': 'gen98617@PCNAME', 'ignore_result': False, 'replaced_task_nesting': 0, 'stamped_headers': None, 'stamps': {}, 'properties': {'correlation_id': '642cb8dc-da32-4061-bbe8-2870f7bdd5c9', 'reply_to': '8df5f54e-6ddb-32b3-a3e1-2fc2991d457e', 'delivery_mode': 2, 'delivery_info': {'exchange': '', 'routing_key': 'test-queue'}, 'priority': 0, 'body_encoding': 'base64', 'delivery_tag': '019530ba-0eb8-7ce7-bd82-09c905872813'}, 'reply_to': '8df5f54e-6ddb-32b3-a3e1-2fc2991d457e', 'correlation_id': '642cb8dc-da32-4061-bbe8-2870f7bdd5c9', 'hostname': 'celery@PCNAME', 'delivery_info': {'exchange': '', 'routing_key':... kwargs:{})
[2025-02-23 11:56:07,271: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:07,271: DEBUG/ForkPoolWorker-2] テストタスクが実行されました
[2025-02-23 11:56:07,273: INFO/ForkPoolWorker-2] Task djangoproject.tasks.test_task[642cb8dc-da32-4061-bbe8-2870f7bdd5c9] succeeded in 0.0025518749898765236s: None
[2025-02-23 11:56:07,347: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "DELETE /v1/queues/test-queue/messages/019530b9-ddfd-7520-93f0-8d22b8b0e616 HTTP/1.1" 200 20
[2025-02-23 11:56:07,350: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:07,428: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "DELETE /v1/queues/test-queue/messages/019530ba-0eb8-7ce7-bd82-09c905872813 HTTP/1.1" 200 20
[2025-02-23 11:56:07,430: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:07,508: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "GET /v1/queues/test-queue/messages HTTP/1.1" 200 1507
[2025-02-23 11:56:07,509: INFO/MainProcess] Task djangoproject.tasks.test_task[4773ee11-84d7-4520-9d13-3ee598942203] received
[2025-02-23 11:56:07,510: DEBUG/MainProcess] TaskPool: Apply <function fast_trace_task at 0x103e8aac0> (args:('djangoproject.tasks.test_task', '4773ee11-84d7-4520-9d13-3ee598942203', {'lang': 'py', 'task': 'djangoproject.tasks.test_task', 'id': '4773ee11-84d7-4520-9d13-3ee598942203', 'shadow': None, 'eta': None, 'expires': None, 'group': None, 'group_index': None, 'retries': 0, 'timelimit': [None, None], 'root_id': '4773ee11-84d7-4520-9d13-3ee598942203', 'parent_id': None, 'argsrepr': '()', 'kwargsrepr': '{}', 'origin': 'gen98617@PCNAME', 'ignore_result': False, 'replaced_task_nesting': 0, 'stamped_headers': None, 'stamps': {}, 'properties': {'correlation_id': '4773ee11-84d7-4520-9d13-3ee598942203', 'reply_to': '8df5f54e-6ddb-32b3-a3e1-2fc2991d457e', 'delivery_mode': 2, 'delivery_info': {'exchange': '', 'routing_key': 'test-queue'}, 'priority': 0, 'body_encoding': 'base64', 'delivery_tag': '019530ba-13a2-75a5-a806-040ab5d53441'}, 'reply_to': '8df5f54e-6ddb-32b3-a3e1-2fc2991d457e', 'correlation_id': '4773ee11-84d7-4520-9d13-3ee598942203', 'hostname': 'celery@PCNAME', 'delivery_info': {'exchange': '', 'routing_key':... kwargs:{})
[2025-02-23 11:56:07,511: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:07,512: DEBUG/ForkPoolWorker-3] テストタスクが実行されました
[2025-02-23 11:56:07,513: INFO/ForkPoolWorker-3] Task djangoproject.tasks.test_task[4773ee11-84d7-4520-9d13-3ee598942203] succeeded in 0.0018645829986780882s: None
[2025-02-23 11:56:07,589: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "GET /v1/queues/test-queue/messages HTTP/1.1" 200 1507
[2025-02-23 11:56:07,590: INFO/MainProcess] Task djangoproject.tasks.test_task[260acf17-543d-4c07-8083-bd7ec1f4f01c] received
[2025-02-23 11:56:07,590: DEBUG/MainProcess] TaskPool: Apply <function fast_trace_task at 0x103e8aac0> (args:('djangoproject.tasks.test_task', '260acf17-543d-4c07-8083-bd7ec1f4f01c', {'lang': 'py', 'task': 'djangoproject.tasks.test_task', 'id': '260acf17-543d-4c07-8083-bd7ec1f4f01c', 'shadow': None, 'eta': None, 'expires': None, 'group': None, 'group_index': None, 'retries': 0, 'timelimit': [None, None], 'root_id': '260acf17-543d-4c07-8083-bd7ec1f4f01c', 'parent_id': None, 'argsrepr': '()', 'kwargsrepr': '{}', 'origin': 'gen98617@PCNAME', 'ignore_result': False, 'replaced_task_nesting': 0, 'stamped_headers': None, 'stamps': {}, 'properties': {'correlation_id': '260acf17-543d-4c07-8083-bd7ec1f4f01c', 'reply_to': '8df5f54e-6ddb-32b3-a3e1-2fc2991d457e', 'delivery_mode': 2, 'delivery_info': {'exchange': '', 'routing_key': 'test-queue'}, 'priority': 0, 'body_encoding': 'base64', 'delivery_tag': '019530ba-066c-7583-8fad-7f49a9f13426'}, 'reply_to': '8df5f54e-6ddb-32b3-a3e1-2fc2991d457e', 'correlation_id': '260acf17-543d-4c07-8083-bd7ec1f4f01c', 'hostname': 'celery@PCNAME', 'delivery_info': {'exchange': '', 'routing_key':... kwargs:{})
[2025-02-23 11:56:07,592: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:07,593: DEBUG/ForkPoolWorker-4] テストタスクが実行されました
[2025-02-23 11:56:07,597: INFO/ForkPoolWorker-4] Task djangoproject.tasks.test_task[260acf17-543d-4c07-8083-bd7ec1f4f01c] succeeded in 0.0045628749940078706s: None
[2025-02-23 11:56:07,662: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "DELETE /v1/queues/test-queue/messages/019530ba-13a2-75a5-a806-040ab5d53441 HTTP/1.1" 200 20
[2025-02-23 11:56:07,664: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:07,729: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "DELETE /v1/queues/test-queue/messages/019530ba-066c-7583-8fad-7f49a9f13426 HTTP/1.1" 200 20
[2025-02-23 11:56:07,731: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:07,807: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "GET /v1/queues/test-queue/messages HTTP/1.1" 200 1507
[2025-02-23 11:56:07,809: INFO/MainProcess] Task djangoproject.tasks.test_task[95495221-d89f-4b55-bd86-64c3d9d02f37] received
[2025-02-23 11:56:07,809: DEBUG/MainProcess] TaskPool: Apply <function fast_trace_task at 0x103e8aac0> (args:('djangoproject.tasks.test_task', '95495221-d89f-4b55-bd86-64c3d9d02f37', {'lang': 'py', 'task': 'djangoproject.tasks.test_task', 'id': '95495221-d89f-4b55-bd86-64c3d9d02f37', 'shadow': None, 'eta': None, 'expires': None, 'group': None, 'group_index': None, 'retries': 0, 'timelimit': [None, None], 'root_id': '95495221-d89f-4b55-bd86-64c3d9d02f37', 'parent_id': None, 'argsrepr': '()', 'kwargsrepr': '{}', 'origin': 'gen98617@PCNAME', 'ignore_result': False, 'replaced_task_nesting': 0, 'stamped_headers': None, 'stamps': {}, 'properties': {'correlation_id': '95495221-d89f-4b55-bd86-64c3d9d02f37', 'reply_to': '8df5f54e-6ddb-32b3-a3e1-2fc2991d457e', 'delivery_mode': 2, 'delivery_info': {'exchange': '', 'routing_key': 'test-queue'}, 'priority': 0, 'body_encoding': 'base64', 'delivery_tag': '019530ba-11af-78ea-a11f-9ffe9011c412'}, 'reply_to': '8df5f54e-6ddb-32b3-a3e1-2fc2991d457e', 'correlation_id': '95495221-d89f-4b55-bd86-64c3d9d02f37', 'hostname': 'celery@PCNAME', 'delivery_info': {'exchange': '', 'routing_key':... kwargs:{})
[2025-02-23 11:56:07,810: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:07,811: DEBUG/ForkPoolWorker-5] テストタスクが実行されました
[2025-02-23 11:56:07,813: INFO/ForkPoolWorker-5] Task djangoproject.tasks.test_task[95495221-d89f-4b55-bd86-64c3d9d02f37] succeeded in 0.002412707981420681s: None
[2025-02-23 11:56:07,872: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "GET /v1/queues/test-queue/messages HTTP/1.1" 200 34
[2025-02-23 11:56:08,035: DEBUG/MainProcess] Timer wake-up! Next ETA 0.9984511669899803 secs.
[2025-02-23 11:56:08,878: DEBUG/MainProcess] Starting new HTTPS connection (1): simplemq.tk1b.api.sacloud.jp:443
[2025-02-23 11:56:08,936: DEBUG/MainProcess] https://simplemq.tk1b.api.sacloud.jp:443 "GET /v1/queues/test-queue/messages HTTP/1.1" 200 34
```