2

メッセージが正しくキューに入れられていないようです。

私は Django を Celery と Kombu と共に使用して、Django 独自のデータベースを Broker Backend として利用しています。必要なのは、非常に単純な Pub/Sub セットアップだけです。最終的には Heroku にデプロイされるので、ローカルで実行するために foreman を使用しています。関連するコードと情報は次のとおりです。

ピップフリーズ

Django==1.4.2
celery==3.0.15
django-celery==3.0.11
kombu==2.5.6

プロフィール

web: source bin/activate; python manage.py run_gunicorn -b 0.0.0.0:$PORT -w 4; python manage.py syncdb
celeryd: python manage.py celeryd -E -B --loglevel=INFO

設定.py

# Celery configuration
import djcelery
CELERY_IMPORTS = ("api.tasks",)
BROKER_URL = "django://localhost//"
djcelery.setup_loader()

put_message

with Connection(settings.BROKER_URL) as conn:
  queue = conn.SimpleQueue('celery')
  queue.put(id)
  queue.close()

API/タスク.py

@task()
def process_next_task():
  with Connection(settings.BROKER_URL) as conn:
    queue = conn.SimpleQueue('celery')
    message = queue.get(block=True, timeout=1)
    id = int(message.payload)
    try:
      Model.objects.get(id=id)
    except Model.DoesNotExist:
      message.reject()
    else:
      # Do stuff here
      message.ack()
    queue.close()

ターミナルでは、foreman start問題なく動作し、次のように表示されます。

started with pid 31835
17:08:22 celeryd.1 | started with pid 31836
17:08:22 web.1     | /usr/local/foreman/bin/foreman-runner: line 41: exec: source: not found
17:08:22 web.1     | 2013-02-14 17:08:22 [31838] [INFO] Starting gunicorn 0.16.1
17:08:22 web.1     | 2013-02-14 17:08:22 [31838] [INFO] Listening at: http://0.0.0.0:5000 (31838)
17:08:22 web.1     | 2013-02-14 17:08:22 [31838] [INFO] Using worker: sync
17:08:22 web.1     | 2013-02-14 17:08:22 [31843] [INFO] Booting worker with pid: 31843
17:08:22 web.1     | 2013-02-14 17:08:22 [31844] [INFO] Booting worker with pid: 31844
17:08:22 web.1     | 2013-02-14 17:08:22 [31845] [INFO] Booting worker with pid: 31845
17:08:22 web.1     | 2013-02-14 17:08:22 [31846] [INFO] Booting worker with pid: 31846
17:08:22 celeryd.1 | [2013-02-14 17:08:22,858: INFO/Beat] Celerybeat: Starting...
17:08:22 celeryd.1 | [2013-02-14 17:08:22,870: WARNING/MainProcess] celery@myhost.local ready.
17:08:22 celeryd.1 | [2013-02-14 17:08:22,873: INFO/MainProcess] consumer: Connected to django://localhost//.
17:08:42 celeryd.1 | [2013-02-14 17:08:42,926: WARNING/MainProcess] Received and deleted unknown message. Wrong destination?!?
17:08:42 celeryd.1 | The full contents of the message body was: body: 25 (2b) {content_type:u'application/json' content_encoding:u'utf-8' delivery_info:{u'priority': 0, u'routing_key': u'celery', u'exchange': u'celery'}}

これらの最後の 2 行はすぐには表示されませんが、上記のput_messageセクションのコードを実行する POST 要求を API が受信すると表示されます。Kombu の完全に拡張された Producer クラスと Consumer クラスを使用して実験しましたが、同じ結果が得られました。

Kombu の SimpleQueue の例: http://kombu.readthedocs.org/en/latest/userguide/examples.html#hello-world-example
Celery Docs: http://docs.celeryproject.org/en/latest/index.html

何か案は？

編集済み

--loglevel=DEBUGprocfile 内で変更すると、ターミナル出力が次のように変更されます。

08:54:33 celeryd.1 | started with pid 555
08:54:33 web.1     | started with pid 554
08:54:33 web.1     | /usr/local/foreman/bin/foreman-runner: line 41: exec: source: not found
08:54:36 web.1     | 2013-02-15 08:54:36 [557] [INFO] Starting gunicorn 0.16.1
08:54:36 web.1     | 2013-02-15 08:54:36 [557] [INFO] Listening at: http://0.0.0.0:5000 (557)
08:54:36 web.1     | 2013-02-15 08:54:36 [557] [INFO] Using worker: sync
08:54:36 web.1     | 2013-02-15 08:54:36 [564] [INFO] Booting worker with pid: 564
08:54:36 web.1     | 2013-02-15 08:54:36 [565] [INFO] Booting worker with pid: 565
08:54:36 web.1     | 2013-02-15 08:54:36 [566] [INFO] Booting worker with pid: 566
08:54:36 web.1     | 2013-02-15 08:54:36 [567] [INFO] Booting worker with pid: 567
08:54:37 celeryd.1 | [2013-02-15 08:54:37,480: DEBUG/MainProcess] [Worker] Loading modules.
08:54:37 celeryd.1 | [2013-02-15 08:54:37,484: DEBUG/MainProcess] [Worker] Claiming components.
08:54:37 celeryd.1 | [2013-02-15 08:54:37,484: DEBUG/MainProcess] [Worker] Building boot step graph.
08:54:37 celeryd.1 | [2013-02-15 08:54:37,484: DEBUG/MainProcess] [Worker] New boot order: {ev, queues, beat, pool, mediator, autoreloader, timers, state-db, autoscaler, consumer}
08:54:37 celeryd.1 | [2013-02-15 08:54:37,489: DEBUG/MainProcess] Starting celery.beat._Process...
08:54:37 celeryd.1 | [2013-02-15 08:54:37,490: DEBUG/MainProcess] celery.beat._Process OK!
08:54:37 celeryd.1 | [2013-02-15 08:54:37,491: DEBUG/MainProcess] Starting celery.concurrency.processes.TaskPool...
08:54:37 celeryd.1 | [2013-02-15 08:54:37,491: INFO/Beat] Celerybeat: Starting...
08:54:37 celeryd.1 | [2013-02-15 08:54:37,506: DEBUG/MainProcess] celery.concurrency.processes.TaskPool OK!
08:54:37 celeryd.1 | [2013-02-15 08:54:37,507: DEBUG/MainProcess] Starting celery.worker.mediator.Mediator...
08:54:37 celeryd.1 | [2013-02-15 08:54:37,507: DEBUG/MainProcess] celery.worker.mediator.Mediator OK!
08:54:37 celeryd.1 | [2013-02-15 08:54:37,507: DEBUG/MainProcess] Starting celery.worker.consumer.BlockingConsumer...
08:54:37 celeryd.1 | [2013-02-15 08:54:37,508: WARNING/MainProcess] celery@myhost.local ready.
08:54:37 celeryd.1 | [2013-02-15 08:54:37,508: DEBUG/MainProcess] consumer: Re-establishing connection to the broker...
08:54:37 celeryd.1 | [2013-02-15 08:54:37,510: INFO/MainProcess] consumer: Connected to django://localhost//.
08:54:37 celeryd.1 | [2013-02-15 08:54:37,628: DEBUG/Beat] Current schedule:
08:54:37 celeryd.1 | <Entry: celery.backend_cleanup celery.backend_cleanup() {<crontab: * 4 * * * (m/h/d/dM/MY)>}
08:54:37 celeryd.1 | [2013-02-15 08:54:37,629: DEBUG/Beat] Celerybeat: Ticking with max interval->5.00 minutes
08:54:37 celeryd.1 | [2013-02-15 08:54:37,658: DEBUG/Beat] Celerybeat: Waking up in 5.00 minutes.
08:54:38 celeryd.1 | [2013-02-15 08:54:38,110: DEBUG/MainProcess] consumer: basic.qos: prefetch_count->16
08:54:38 celeryd.1 | [2013-02-15 08:54:38,126: DEBUG/MainProcess] consumer: Ready to accept tasks!
08:55:08 celeryd.1 | [2013-02-15 08:55:08,184: WARNING/MainProcess] Received and deleted unknown message. Wrong destination?!?
08:55:08 celeryd.1 | The full contents of the message body was: body: 26 (2b) {content_type:u'application/json' content_encoding:u'utf-8' delivery_info:{u'priority': 0, u'routing_key': u'celery', u'exchange': u'celery'}}
4

4 に答える 4

3

問題は 2 つあります。

メッセージ形式が間違っていました。@asksol が提供したhttp://docs.celeryproject.org/en/latest/internals/protocol.htmlのドキュメントに従って辞書である必要があり、そのページの下部にある例に従ってください。

メッセージの例

{"id": "4cc7438e-afd4-4f8f-a2f3-f46567e7ca77",
 "task": "celery.task.PingTask",
 "args": [],
 "kwargs": {},
 "retries": 0,
 "eta": "2009-11-17T12:30:56.527191"}

put_message

with Connection(settings.BROKER_URL) as conn:
  queue = conn.SimpleQueue('celery')
  message = {
    'task': 'process-next-task',
    'id': str(uuid.uuid4()),
    'args': [id],
    "kwargs": {},
    "retries": 0,
    "eta": str(datetime.datetime.now())
  }
  queue.put(message)
  queue.close()

Procfile プロセスは、タスクを実行するコンシューマーであるため、タスク内にコンシューマーをセットアップする必要はありません。メッセージを公開したときに送信したパラメーターを使用する必要がありました。

API/タスク.py

@task(serializer='json', name='process-next-task')
def process_next_task(id):
  try:
    Model.objects.get(id=int(id))
  except Model.DoesNotExist:
    pass
  else:
    # Do stuff here
于 2013-02-15T16:19:39.897 に答える