어떻게

Question

내가 셀러리의 test 큐가 task.apply_async 셀러리에 우선 순위를 사용하고 나는 그것을 위해 작업을 정의 appy_async에 대해 here으로 정의된다. 나는이 같은 몇 가지 결과를 볼 수 excpected

for i in range(100): 
    priority_test.apply_async((i%10,), queue="test", priority=i%10) 

: 그래서,이 같은 for loop 썼다 서로 후 동일한 우선 순위를 실행

[2017-12-26 17:21:37,309: WARNING/ForkPoolWorker-1] 10 
[2017-12-26 17:21:37,311: WARNING/ForkPoolWorker-1] 10 
[2017-12-26 17:21:37,314: WARNING/ForkPoolWorker-1] 10 
[2017-12-26 17:21:37,317: WARNING/ForkPoolWorker-1] 9 
[2017-12-26 17:21:37,319: WARNING/ForkPoolWorker-1] 9 
[2017-12-26 17:21:37,321: WARNING/ForkPoolWorker-1] 9 
[2017-12-26 17:21:37,323: WARNING/ForkPoolWorker-1] 8 
[2017-12-26 17:21:37,326: WARNING/ForkPoolWorker-1] 8 
[2017-12-26 17:21:37,329: WARNING/ForkPoolWorker-1] 8 
[2017-12-26 17:21:37,332: WARNING/ForkPoolWorker-1] 7 
[2017-12-26 17:21:37,334: WARNING/ForkPoolWorker-1] 7 
[2017-12-26 17:21:37,336: WARNING/ForkPoolWorker-1] 7 
[2017-12-26 17:21:37,341: WARNING/ForkPoolWorker-1] 6 
[2017-12-26 17:21:37,344: WARNING/ForkPoolWorker-1] 6 
[2017-12-26 17:21:37,346: WARNING/ForkPoolWorker-1] 6 
[2017-12-26 17:21:37,349: WARNING/ForkPoolWorker-1] 5 
[2017-12-26 17:21:37,351: WARNING/ForkPoolWorker-1] 5 
[2017-12-26 17:21:37,353: WARNING/ForkPoolWorker-1] 5 
[2017-12-26 17:21:37,355: WARNING/ForkPoolWorker-1] 4 
[2017-12-26 17:21:37,358: WARNING/ForkPoolWorker-1] 4 
[2017-12-26 17:21:37,360: WARNING/ForkPoolWorker-1] 4 

수단을하지만 일반적인 방법으로이를 실행 :

가 어떻게 rabbitmq와 셀러리에 priority을 적용해야하며, 위의 문서에서 priority 속성은 무엇인가

[2017-12-26 17:21:37,309: WARNING/ForkPoolWorker-1] 10 
[2017-12-26 17:21:37,311: WARNING/ForkPoolWorker-1] 9 
[2017-12-26 17:21:37,314: WARNING/ForkPoolWorker-1] 8 
[2017-12-26 17:21:37,317: WARNING/ForkPoolWorker-1] 7 
[2017-12-26 17:21:37,319: WARNING/ForkPoolWorker-1] 6 
[2017-12-26 17:21:37,321: WARNING/ForkPoolWorker-1] 5 
[2017-12-26 17:21:37,323: WARNING/ForkPoolWorker-1] 4 
[2017-12-26 17:21:37,326: WARNING/ForkPoolWorker-1] 3 
[2017-12-26 17:21:37,329: WARNING/ForkPoolWorker-1] 2 
[2017-12-26 17:21:37,332: WARNING/ForkPoolWorker-1] 1 
[2017-12-26 17:21:37,334: WARNING/ForkPoolWorker-1] 10 
[2017-12-26 17:21:37,336: WARNING/ForkPoolWorker-1] 9 
[2017-12-26 17:21:37,341: WARNING/ForkPoolWorker-1] 8 
[2017-12-26 17:21:37,344: WARNING/ForkPoolWorker-1] 7 
[2017-12-26 17:21:37,346: WARNING/ForkPoolWorker-1] 6 
[2017-12-26 17:21:37,349: WARNING/ForkPoolWorker-1] 5 
[2017-12-26 17:21:37,351: WARNING/ForkPoolWorker-1] 4 
[2017-12-26 17:21:37,353: WARNING/ForkPoolWorker-1] 3 
[2017-12-26 17:21:37,355: WARNING/ForkPoolWorker-1] 2 
[2017-12-26 17:21:37,358: WARNING/ForkPoolWorker-1] 1 
[2017-12-26 17:21:37,360: WARNING/ForkPoolWorker-1] 10 
[2017-12-26 17:21:37,362: WARNING/ForkPoolWorker-1] 9 
[2017-12-26 17:21:37,364: WARNING/ForkPoolWorker-1] 8 
[2017-12-26 17:21:37,365: WARNING/ForkPoolWorker-1] 7 
[2017-12-26 17:21:37,367: WARNING/ForkPoolWorker-1] 6 
[2017-12-26 17:21:37,369: WARNING/ForkPoolWorker-1] 5 
[2017-12-26 17:21:37,371: WARNING/ForkPoolWorker-1] 4 
[2017-12-26 17:21:37,373: WARNING/ForkPoolWorker-1] 3 
[2017-12-26 17:21:37,374: WARNING/ForkPoolWorker-1] 2 
[2017-12-26 17:21:37,376: WARNING/ForkPoolWorker-1] 1 

?

Answer 1

priority 제대로 작동하려면 몇 가지 설정을 올바르게 구성해야하며 RabbitMQ 버전 3.5.0 이상이 필요합니다.

모든 큐의 기본 값은 task_queue_max_priority 설정을 사용하여 설정할 수 있습니다

from kombu import Exchange, Queue 

app.conf.task_queues = [ 
    Queue('tasks', Exchange('tasks'), routing_key='tasks', 
      queue_arguments={'x-max-priority': 10}, 
] 

:

app.conf.task_queue_max_priority = 10 

그런 다음 다음을 구성을

첫 번째는 docs에서 (10)에 큐의 x-max-priority 설정 설정 :

CELERY_ACKS_LATE = True 
CELERYD_PREFETCH_MULTIPLIER = 1 

기본적으로 프리 페치 승수는 4입니다.이 경우 다른 작업이 대기열에 나타나기 전에 우선 순위가 10, 9, 8 및 7 인 처음 4 개 작업을 가져옵니다. CELERY_ACKS_LATE 설정은 작업이 실행 된 후 승인되도록합니다. 이 설정으로 실험하여 원하는 동작을 확인할 수 있습니다.

Answer 2

나는 비트

예를 들어

사용 셀러리 내 작업을 예약해야합니다 내가 하나 개의 작업 (1) 5 초에 대한 일정을 10 초 (2) 두 번째 작업을 예약을하지만 난이었다

5 (1 일): 일정

출력 asynchronus 점점10 (2 일) 5 (1 일) 5 (1 일) 5 (1 일) 10 (2 일)

은이 같은 출력을 필요 :

출력 :

5 (1) 5 (1) 10 (2) 5 (1) 5 (1) 10 (2) 5 (1) 5 (1) 10 (2) 5 (1) 5 (1) 10 (2)

매 5 초 작업 (1) O/P 관련된 다음 작업 (2) 연산 10 초

# celery worker --loglevel=info -A tasks --beat 
import datetime 
from celery import Celery 
from celery import group 

import datetime 
from celery import Celery 
from celery import group 

app=Celery('tasks',broker= 'pyamqp://guest@localhost//', 
backend='db+postgresql://postgres:newpassword@localhost/postgres') 

@app.task 
def first(string): 
    print(string) 

@app.task 
def second(string): 
    print(string) 


app.conf.update(
    beat_schedule= { 
     "multiply-each-05-seconds": { 
     "task": "tasks.first", 
     "schedule": 05.0, 
     "args": ('I am first...OOO',) 
    }, 
    'add-every-10-seconds': { 
     'task': 'tasks.second', 
     'schedule': 10.0, 
     'args': ('You are Second haha..!!',) 
    }, 
    }, 
) 

# sudo apt-get install recordmydesktop 
# recordmydesktop --no-sound 

01,235,164 때문에

---