Thông báo cho công việc cần tây của công nhân tắt máy

Question

Tôi đang sử dụng cần tây 2.4.1 với python 2.6, phần phụ trợ thỏmq và django. Tôi muốn công việc của tôi có thể dọn dẹp đúng cách nếu nhân viên đóng cửa. Theo tôi biết bạn không thể cung cấp một destructor nhiệm vụ vì vậy tôi đã cố gắng hooking vào tín hiệu worker_shutdown.

Lưu ý: AbortableTask chỉ hoạt động với phần phụ trợ cơ sở dữ liệu để tôi không thể sử dụng.

from celery.signals import worker_shutdown 

@task 
def mytask(*args) 

    obj = DoStuff() 

    def shutdown_hook(*args): 
    print "Worker shutting down" 
    # cleanup nicely 
    obj.stop() 

    worker_shutdown.connect(shutdown_hook) 

    # blocking call that monitors a network connection 
    obj.stuff() 

Tuy nhiên, móc tắt máy không bao giờ được gọi. Ctrl-C'ing người lao động không giết nhiệm vụ và tôi phải tự giết nó khỏi trình bao.

Vì vậy, nếu đây không phải là cách thích hợp để thực hiện, làm cách nào để cho phép các tác vụ tắt trơn tru?

Answer 1

worker_shutdown chỉ được gửi bởi MainProcess, không phải công nhân trẻ em trong hồ bơi. Tất cả các tín hiệu worker_*except for worker_process_init, tham khảo MainProcess.

Tuy nhiên, móc tắt máy không bao giờ được gọi. Ctrl-C'ing các công nhân không giết nhiệm vụ và tôi phải tự giết nó khỏi vỏ.

Công nhân không bao giờ ngừng hoạt động khi tắt máy (bình thường). Ngay cả khi một công việc mất nhiều ngày để hoàn thành, công nhân sẽ không hoàn thành tắt máy cho đến khi hoàn thành. Bạn có thể đặt --soft-time-limit hoặc --time-limit thành để cho biết ví dụ khi nào là ok để chấm dứt tác vụ.

Vì vậy, để thêm bất kỳ loại quy trình dọn dẹp quy trình nào, trước hết bạn cần phải đảm bảo rằng các tác vụ có thể thực sự hoàn tất. Khi dọn dẹp sẽ không được gọi trước khi điều đó xảy ra.

Để thêm một bước dọn dẹp cho người lao động hồ xử lý, bạn có thể sử dụng cái gì đó như:

from celery import platforms 
from celery.signals import worker_process_init 

def cleanup_after_tasks(signum, frame): 
    # reentrant code here (see http://docs.python.org/library/signal.html) 

def install_pool_process_sighandlers(**kwargs): 
    platforms.signals["TERM"] = cleanup_after_tasks 
    platforms.signals["INT"] = cleanup_after_tasks 

worker_process_init.connect(install_pool_process_sighandlers)