13

セロリ労働者のリアルタイム監視に関するセロリチュートリアルによると、労働者によって生成されたイベントをプログラムでキャプチャし、それに応じてアクションを実行することもできます。

私の質問は、Celery-Djangoアプリケーションで、この例のモニターをどのように統合できるかということです。

編集：チュートリアルのコード例は次のようになります。

from celery import Celery

def my_monitor(app):
    state = app.events.State()

    def announce_failed_tasks(event):
        state.event(event)
        task_id = event['uuid']

        print('TASK FAILED: %s[%s] %s' % (
            event['name'], task_id, state[task_id].info(), ))
    with app.connection() as connection:
        recv = app.events.Receiver(connection, handlers={
                'task-failed': announce_failed_tasks,
                'worker-heartbeat': announce_dead_workers,
        })
        recv.capture(limit=None, timeout=None, wakeup=True)

if __name__ == '__main__':
    celery = Celery(broker='amqp://guest@localhost//')
    my_monitor(celery)

したがって、ワーカーによって送信されたtask_failedイベントをキャプチャし、チュートリアルが示すようにそのtask_idを取得して、アプリケーション用に構成された結果バックエンドからこのタスクの結果を取得し、さらに処理したいと思います。私の問題は、アプリケーションの入手方法がはっきりしないことです。django-celeryプロジェクトでは、Celeryライブラリのインスタンス化が透過的ではないためです。

また、ワーカーがタスクの実行を終了したときに結果を処理する方法について、他のアイデアも受け入れています。

4

2 に答える 2

19

わかりました、これが解決策であるかどうかはわかりませんが、これを行う方法を見つけましたが、それは私にとってはうまくいきます。モニター機能は基本的にブローカーに直接接続し、さまざまなタイプのイベントをリッスンします。私のコードは次のようになります。

from celery.events import EventReceiver
from kombu import Connection as BrokerConnection

def my_monitor:
    connection = BrokerConnection('amqp://guest:guest@localhost:5672//')

    def on_event(event):
        print "EVENT HAPPENED: ", event

    def on_task_failed(event):
        exception = event['exception']
        print "TASK FAILED!", event, " EXCEPTION: ", exception

    while True:
        try:
            with connection as conn:
                recv = EventReceiver(conn,
                                 handlers={'task-failed' : on_task_failed,
                                           'task-succeeded' : on_event,
                                           'task-sent' : on_event,
                                           'task-received' : on_event,
                                           'task-revoked' : on_event,
                                           'task-started' : on_event,
                                           # OR: '*' : on_event
                                           })
            recv.capture(limit=None, timeout=None)
    except (KeyboardInterrupt, SystemExit):
        print "EXCEPTION KEYBOARD INTERRUPT"
        sys.exit()

これですべてです。そして、これを通常のアプリケーションとは異なるプロセスで実行します。つまり、この関数のみを実行するセロリアプリケーションの子プロセスを作成します。HTH

于 2013-03-19T11:46:31.093 に答える
8

いくつかの落とし穴に注意してください

  1. CELERY_SEND_EVENTSセロリ設定でフラグをtrueに設定する必要があります。
  2. ワーカーからの新しいスレッドでイベントモニターを設定することもできます。

これが私の実装です：

class MonitorThread(object):
    def __init__(self, celery_app, interval=1):
        self.celery_app = celery_app
        self.interval = interval

        self.state = self.celery_app.events.State()

        self.thread = threading.Thread(target=self.run, args=())
        self.thread.daemon = True
        self.thread.start()

    def catchall(self, event):
        if event['type'] != 'worker-heartbeat':
            self.state.event(event)

        # logic here

    def run(self):
        while True:
            try:
                with self.celery_app.connection() as connection:
                    recv = self.celery_app.events.Receiver(connection, handlers={
                        '*': self.catchall
                    })
                    recv.capture(limit=None, timeout=None, wakeup=True)

            except (KeyboardInterrupt, SystemExit):
                raise

            except Exception:
                # unable to capture
                pass

            time.sleep(self.interval)

if __name__ == '__main__':
    app = get_celery_app() # returns app
    MonitorThread(app)
    app.start()
于 2015-09-08T04:24:25.023 に答える