セロリタスクを特定のキューにルーティングする

Question

私のサーバーで実行されている2つの別個のcelerydプロセスがsupervisorで管理されています。彼らは、次のような別々のキューで待機するように設定されています。

[program:celeryd1] 
command=/path/to/celeryd --pool=solo --queues=queue1 
... 

[program:celeryd2] 
command=/path/to/celeryd --pool=solo --queues=queue2 
... 

そして、私のceleryconfigは次のようなものになります。

from celery.schedules import crontab 

BROKER_URL = "amqp://guest:guest@localhost:5672//" 

CELERY_DISABLE_RATE_LIMITS = True 
CELERYD_CONCURRENCY = 1 
CELERY_IGNORE_RESULT = True 

CELERY_DEFAULT_QUEUE = 'default' 
CELERY_QUEUES = { 
    'default': { 
     "exchange": "default", 
     "binding_key": "default", 
    }, 
    'queue1': { 
     'exchange': 'queue1', 
     'routing_key': 'queue1', 
    }, 
    'queue2': { 
     'exchange': 'queue2', 
     'routing_key': 'queue2', 
    }, 
} 

CELERY_IMPORTS = ('tasks',) 

CELERYBEAT_SCHEDULE = { 
    'first-queue': { 
     'task': 'tasks.sync', 
     'schedule': crontab(hour=02, minute=00), 
     'kwargs': {'client': 'client_1'}, 
     'options': {'queue': 'queue1'}, 
    }, 
    'second-queue': { 
     'task': 'tasks.sync', 
     'schedule': crontab(hour=02, minute=00), 
     'kwargs': {'client': 'client_2'}, 
     'options': {'queue': 'queue1'}, 
    }, 
} 

すべてtasks.syncタスクは、特定のキュー（したがってceleryd進行）にルーティングする必要があります。しかし、私がsync.apply_async(kwargs={'client': 'value'}, queue='queue1')で手動でタスクを実行しようとすると、両方のセロリの労働者が仕事を引き受けます。タスクを正しいキューにルーティングするにはどうしたらよいですか？キューにバインドされているワーカーによってのみ実行されますか？

Answer 1

セロビートインスタンスは1つだけ実行されていますか？

多分あなたはこれと衝突する古いキューバインディングを持っていますか？ rabbitmqctl list_queuesとrabbitmqctl list_bindingsを実行してみてください。 ブローカ内のデータを一からリセットすることがあります。

あなたがここにある例はうまくいくはずです。私が試したときに私のために働いています。

ヒント：キュー名と同じexchangeおよびbinding_key値を使用しているため、 を明示的にCELERY_QUEUESにリストする必要はありません。 CELERY_CREATE_MISSING_QUEUES がオンの場合（既定値では）、celeryd -Q queue1の場合は のようにキューが自動的に作成され、未定義のキューにはタスクが送信されます。