Dừng/Thanh toán các công việc định kỳ ở Django-Celery

Question

Tôi đã quản lý để có được các tác vụ định kỳ làm việc trong django-celery bằng cách phân lớp PeriodicTask. Tôi đã cố gắng tạo ra một nhiệm vụ thử nghiệm và đặt nó chạy làm một cái gì đó vô ích. Nó hoạt động.

Bây giờ tôi không thể dừng lại. Tôi đã đọc tài liệu và tôi không thể tìm hiểu cách xóa nhiệm vụ khỏi hàng đợi thực hiện. Tôi đã thử sử dụng celeryctl và sử dụng shell, nhưng registry.tasks() là rỗng, vì vậy tôi không thể xem làm thế nào để loại bỏ nó.

Tôi đã thấy các đề xuất rằng tôi nên "thu hồi" nó, nhưng đối với điều này tôi dường như cần một id công việc và tôi không thể thấy cách tôi sẽ tìm thấy id tác vụ.

Cảm ơn.

Answer 1

Tác vụ là một tin nhắn và "tác vụ định kỳ" gửi thông báo tác vụ theo các khoảng thời gian định kỳ. Mỗi tác vụ được gửi sẽ có một id duy nhất được gán cho nó.

revoke sẽ chỉ hủy một thông báo tác vụ duy nhất. Để lấy id cho tác vụ, bạn phải giữ bản nhạc của id được gửi, nhưng bạn cũng có thể chỉ định id tùy chỉnh khi bạn gửi tác vụ.

Tôi không chắc chắn nếu bạn muốn hủy một thông báo tác vụ duy nhất, hoặc nếu bạn muốn dừng tác vụ định kỳ gửi nhiều tin nhắn hơn, vì vậy tôi sẽ liệt kê các câu trả lời cho cả hai.

Không có cách tích hợp để giữ id của tác vụ được gửi với nhiệm vụ định kỳ, nhưng bạn có thể đặt id cho mỗi tác vụ thành tên của tác vụ định kỳ, theo cách đó id sẽ tham chiếu đến bất kỳ nhiệm vụ được gửi với nhiệm vụ định kỳ (thường là tác vụ cuối cùng). Bạn có thể chỉ định một id tùy chỉnh theo cách này,

hoặc với @periodic_task trang trí:

@periodic_task(options={"task_id": "my_periodic_task"}) 
def my_periodic_task(): 
    pass 

hoặc với các thiết lập CELERYBEAT_SCHEDULE:

CELERYBEAT_SCHEDULE = {name: {"task": task_name, 
           "options": {"task_id": name}}} 

Nếu bạn muốn loại bỏ một công việc định kỳ bạn chỉ cần loại bỏ các @periodic_task từ codebase, hoặc loại bỏ các mục từ CELERYBEAT_SCHEDULE. Nếu bạn đang sử dụng bộ lập lịch cơ sở dữ liệu Django, bạn phải xóa nhiệm vụ định kỳ khỏi giao diện Quản trị Django.

PS1: revoke không dừng tác vụ đã được bắt đầu. Nó chỉ hủy bỏ các nhiệm vụ chưa được bắt đầu. Bạn có thể chấm dứt tác vụ đang chạy bằng cách sử dụng revoke(task_id, terminate=True). Theo mặc định, điều này sẽ gửi tín hiệu TERM tới quy trình, nếu bạn muốn gửi một tín hiệu khác (ví dụ: KILL) sử dụng revoke(task_id, terminate=True, signal="KILL").

PS2: thu hồi là lệnh điều khiển từ xa nên nó chỉ được hỗ trợ bởi giao thông vận tải RabbitMQ và Redis. Nếu bạn muốn công việc của bạn để hỗ trợ hủy bạn nên làm như vậy bằng cách lưu trữ một lá cờ cancelled trong một cơ sở dữ liệu và có nhiệm vụ kiểm tra cờ khi nó bắt đầu:

from celery.task import Task 

class RevokeableTask(Task): 
    """Task that can be revoked. 

    Example usage: 

     @task(base=RevokeableTask) 
     def mytask(): 
      pass 
    """ 

    def __call__(self, *args, **kwargs): 
     if revoke_flag_set_in_db_for(self.request.id): 
      return 
     super(RevokeableTask, self).__call__(*args, **kwargs) 

Answer 2

Chỉ trong trường hợp này có thể giúp ai đó ... Chúng tôi đã có cùng một vấn đề trong công việc, và khinh thường một số nỗ lực để tìm thấy một số loại lệnh quản lý để loại bỏ các nhiệm vụ định kỳ, chúng tôi có thể không. Vì vậy, đây là một số gợi ý.

Trước tiên bạn có thể nên kiểm tra kỹ xem scheduler class nào bạn đang sử dụng.

Trình lên lịch mặc định là celery.beat.PersistentScheduler, đơn giản là theo dõi thời gian chạy cuối cùng trong tệp cơ sở dữ liệu cục bộ (giá treo).

Trong trường hợp của chúng tôi, chúng tôi đang sử dụng lớp học djcelery.schedulers.DatabaseScheduler.

django-cần tây cũng tàu với một lịch trình lưu trữ lịch trình trong cơ sở dữ liệu Django

Mặc dù tài liệu nào đề cập đến một cách để loại bỏ các nhiệm vụ tuần hoàn:

Sử dụng django-celery của lịch trình bạn có thể thêm, sửa đổi và loại bỏ các nhiệm vụ định kỳ từ quản trị Django.

Chúng tôi muốn thực hiện việc xóa theo chương trình hoặc thông qua lệnh (cần tây/quản lý) trong trình bao.

Vì chúng ta không thể tìm thấy một dòng lệnh, chúng tôi sử dụng vỏ django/python:

$ python manage.py shell 
>>> from djcelery.models import PeriodicTask 
>>> pt = PeriodicTask.objects.get(name='the_task_name') 
>>> pt.delete() 

Tôi hy vọng điều này sẽ giúp!