Làm thế nào để gửi nhiệm vụ tuần hoàn vào hàng đợi cụ thể trong cần tây

Question

Bằng Cần tây mặc định gửi tất cả các nhiệm vụ để 'cần tây' hàng đợi, nhưng bạn có thể thay đổi hành vi này bằng cách bổ sung thêm tham số:

@task(queue='celery_periodic') 
def recalc_last_hour(): 
    log.debug('sending new task') 
    recalc_hour.delay(datetime(2013, 1, 1, 2)) # for example 

thiết lập Scheduler:

CELERYBEAT_SCHEDULE = { 
    'installer_recalc_hour': { 
     'task': 'stats.installer.tasks.recalc_last_hour', 
     'schedule': 15 # every 15 sec for test 
    }, 
} 
CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler" 

Run nhân:

python manage.py celery worker -c 1 -Q celery_periodic -B -E 

chương trình này không hoạt động như mong đợi: công nhân này gửi nhiệm vụ tuần hoàn t o 'cần tây' hàng đợi, không phải 'celery_periodic'. Lam sao tôi co thể sửa no?

P.S. celery == 3.0.16

Answer 1

Tôi tìm thấy giải pháp cho vấn đề này:

1) Trước hết tôi đã thay đổi cách định cấu hình nhiệm vụ định kỳ. Tôi đã từng @periodic_task trang trí như thế này:

@periodic_task(run_every=crontab(minute='5'), 
       queue='celery_periodic', 
       options={'queue': 'celery_periodic'}) 
def recalc_last_hour(): 
    dt = datetime.utcnow() 
    prev_hour = datetime(dt.year, dt.month, dt.day, dt.hour) \ 
       - timedelta(hours=1) 
    log.debug('Generating task for hour %s', str(prev_hour)) 
    recalc_hour.delay(prev_hour) 

2) tôi đã viết celery_periodic hai lần trong params để @periodic_task:

• queue = 'celery_periodic' tùy chọn được sử dụng khi bạn gọi tác vụ từ mã (.delay hoặc .apply_async)

• tùy chọn = {'queue': 'celery_periodic'} tùy chọn được sử dụng khi cần đánh bại gọi nó.

Tôi chắc chắn, điều tương tự cũng có thể xảy ra nếu bạn định cấu hình các tác vụ định kỳ với biến CELERYBEAT_SCHEDULE.

UPD. Giải pháp này đúng cho cả lưu trữ dựa trên DB và dựa trên tệp cho CELERYBEAT_SCHEDULER.

Answer 2

Định kỳ được gửi đến hàng đợi bởi celerybeat.Bạn có thể làm mọi việc chúng tôi làm với Celery api. Dưới đây là danh sách các cấu hình đi kèm với celerybeat.

http://celery.readthedocs.org/en/latest/userguide/periodic-tasks.html#available-fields

Trong trường hợp của bạn

CELERYBEAT_SCHEDULE = { 
    'installer_recalc_hour': { 
     'task': 'stats.installer.tasks.recalc_last_hour', 
     'schedule': 15 # every 15 sec for test, 
     'options': {'queue' : 'celery_periodic'} ##options are mapped to apply_async options 
    }, 
} 

Answer 3

Và nếu bạn đang sử dụng djcelery Cơ sở dữ liệu lịch trình, bạn có thể xác định hàng đợi trên Options Thi Công -> đợi lĩnh vực