据有关芹菜教程芹菜工人的实时监控 ，还可以通过编程捕获由工人产生的事件，并采取相应的行动。
我的问题是如何监控结合起来，作为一个在这个例子中，在芹菜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的芹菜项目是不是透明的，我芹菜库的实例。
我也开放给任何其他想法，当工人已完成执行任务怎么处理结果。
好吧，我发现这样做的方式，但我不知道这是解决方案，但它为我工作。 监视功能基本上是直接连接到代理，并听取不同类型的事件。 我的代码如下所示：
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
当心几个陷阱的
- 您需要设置
CELERY_SEND_EVENTS标志为真在你的芹菜配置。 - 您还可以设置事件监视器从你的工人一个新的线程。
下面是我的实现：
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()
