Блог

celery_tasks.py

 from celery import Celery

app = Celery('tasks')
app.config_from_object('celeryconfig')

@app.task(default_retry_delay=300, max_retries=5, acks_late=True)
def my_task_A(x, y):
    return x   y

if __name__ == '__main__':
    app.start()
  

celeryconfig.py

 from kombu import Exchange, Queue

BROKER_URL = 'amqp://'

CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT=['json']
CELERY_TIMEZONE = 'America/Los_Angeles'
CELERY_ENABLE_UTC = True

CELERY_QUEUES = (
    Queue('default',    Exchange('default'),    routing_key='default'),
    Queue('for_task_A', Exchange('for_task_A'), routing_key='for_task_A'),
    Queue('for_task_B', Exchange('for_task_B'), routing_key='for_task_B'),
)

CELERY_ROUTES = {
    'celery_tasks.my_task_A': {'queue': 'for_task_A', 'routing_key': 'for_task_A'},
    'celery_tasks.my_task_B': {'queue': 'for_task_B', 'routing_key': 'for_task_B'},
}
  

celery_test.py

 from celery_tasks import add2, my_task_A, process_lots_of_items
from celery import group

items = zip(xrange(1000), xrange(1000))

result = my_task_A.chunks(items, 10).delay()
result.save()    
print result
  

Я запускаю рабочий таким образом:

 celery -A celery_tasks worker --concurrency 20 --loglevel=info -Q for_task_A
  

Я получаю это в «результате печати»

 <GroupResult: 446d7f30-100c-4ea6-984c-132583d7cdec [cbf01a75-90c6-4ac7-9a47-9492bdd68908, 19e800e8-7d96-45a7-853c-87e98b1298cb, a94527bf-e29a-4fe5-bfe2-97d80e8fab07, 7a5f2182-2bec-42c2-ac41-394179ddf57e, 60cf80ac-a0a3-45df-8962-2d3603212434, bdae9da8-6488-4292-a2ad-96cf457466d2, b181f852-b54f-4011-a063-e65fc37d4236, c4a45679-7409-4769-b4e2-b05779e41a15, f5fb9a39-1a94-4e3f-9962-0793e7b316bb, 47aea034-e2d2-4600-a47a-37aecd6b9243, c392940a-21cc-48f5-a627-322a81546ebe, b7098e64-dc5c-4eb5-90c6-bc55f1b1c79f, a1ee6d93-1c74-40fb-a838-a913e963995d, 1a442a7e-cce0-434f-8637-7cfb13c6b4f5, b07f5051-765f-4345-82b2-bb752386371f, 68ffbdf6-3e22-4caa-a09f-fe4084841528, 9dd51dad-2794-47d9-b73d-3d3605e23b03, 5f118133-d860-412a-954b-3b54a2cbb192, 76ba100f-e52a-411f-aa29-7f59deb928d8, 074811ac-2d62-49e1-ae66-d16b59655248, 160a9152-2b83-41a2-9dfd-83ed27f50e0b, 82b000f4-9a68-4f12-b450-b74224d8630e, 18d9d1ec-3434-4c73-ba07-40cec3485331, a363d468-4483-4ee9-9406-cb448d5cfb1b, bc825b16-3afd-413b-b289-7447e5edec9b, a1dc62ba-5113-4f2a-b4ef-36c013b0f325, 58a1e397-e729-4640-9acf-d64308dff655, 4699df15-5beb-4920-bfa3-af45d732665c, 24421363-5c5d-4a3e-9046-be965ac174bb, 43065be9-f707-45c6-98e3-c0a4dacc3816, d86f8a4c-6ee7-4190-9c3c-7bbe2df4332c, 9cf60b6c-fc3a-44c6-875b-17cd06c8ffb8, e4c769ef-d457-446c-a4ea-0cfc619cd9f4, 9551b7de-6c8c-4c6a-bb19-f90af6e581b5, 93fd5354-ea7b-4fb1-badd-6641e5d2a151, a2196738-b2e3-42be-9319-eb7f0c669cf2, b70b0a7a-72af-42e5-9369-be1789c7466c, 10adabc7-1471-4b69-b765-e4d629afaded, d5545f86-582c-4995-adfb-3d4f33efdeee, 23356b97-8959-4b5d-b7b6-a3983f31253e, 65113a60-6763-4311-a6da-bdc5ba63210c, 09945059-e23f-458a-be61-80a52035f620, 0976cde4-2a23-41d6-9e6f-80980b72026f, 7593ba47-e350-4a7a-a0f6-27d3a21a993f, 89d1e153-bdc4-4875-b128-c8849ff68445, 013aa255-182d-4c4f-849c-467ef927c3b2, 4d075dd7-1449-4d77-8ef5-1ae75eb085c5, 5a456e8c-befa-4c46-84ce-f04bb0608337, a07a60ae-b8fa-4d8f-9b7d-67b3ba54ff8a, 100ea258-bb91-469c-ad75-94071912710a, 04e92094-e456-4b0e-8485-9a89dd8edf67, 49ffb4b3-02c8-491c-bd3e-711c6796fc89, 4070fbaf-1c7e-4592-a756-35961a506aee, 7629e30c-b96a-4cf7-813b-a873e49d9e1a, ac7abd92-169b-4712-a8cf-0a27544616e2, a8c4777a-78f4-4a54-b461-27e7d085977d, f371698f-c6cb-41d7-8c22-07d006dbc3a5, e0621c5c-f676-42bb-9f7e-01c06bc52956, e9691699-378f-48aa-92a3-15d9838df51e, 8b113521-aa60-4603-9098-833d0dea595c, 3583a382-9d86-44da-aa5f-64e62a166c24, d0df2f06-281c-4252-911c-046ad1a125fc, d736733e-7e87-4217-ba41-a2a13769c225, 24555228-bc85-40ae-a8aa-6c097f58ef69, 4d1a330b-c0eb-4281-af1c-66d2b0709b5f, 097d5578-d773-404f-8d6d-650a85f3ba2a, 71944d43-a294-422e-936b-1a3cf39985e1, 891197d6-d53a-4f0e-b6ca-f3a42a65df5c, c9119d99-37b3-4f60-bb16-fcf7a863e41a, 5f1214fd-31f1-4616-8721-49adb52e477c, 6c833501-0135-4a73-bc7e-725d548b128f, 78f8e3c5-d296-43df-a710-2d91a374a294, 80bf399e-cc06-4884-9b84-923cb9bd4afd, c7fd8dc4-5c95-48e4-9816-6cc7abb8a41a, 24647ebb-6881-4601-9b41-64907cc3d670, 9e110ad1-5312-4b59-a7cc-a7b1a53890a8, a967863e-5892-4b9b-9678-ecbe2d9c6f62, 074222f3-4edf-4f7b-8154-7288f9d27ff6, 0352820c-9aa8-4034-b6ef-e020d34112f6, 1a3f8329-93e4-4a1e-85f2-02559c0d614c, e083f94e-ebee-4f5b-b18f-970be79f8789, ce95a183-512e-4157-b854-51c4b54171d8, f6c5b086-d184-4638-b519-23e04e66001f, e367b288-872e-495f-b111-c0d936d5d44a, 848e1e53-ee88-461d-8de7-9d518f76c2c7, e89a6e6a-84d7-4e2f-8868-cc57dcc3874e, b16fee13-02cd-4037-bcab-c13b244514d9, 86cd3bb7-5ad7-42cf-ba00-9133f8503411, 4bb76f43-149e-4edb-9145-0ce3ab800aeb, 23179c27-55b1-47af-9bc2-5320e63447fe, 07acf561-b0ff-47db-8cdf-0918f41329eb, 2312ee80-cef4-459d-8f41-aea2a05af903, bbab3302-ba79-4134-9264-5e183eacf422, e40a964f-79ba-4a8d-abd9-5d994fc7f1ea, 59da2380-7871-4bbf-8baf-4859f0d154fe, 533a3ae8-3f63-4dab-be74-93a104948fe4, 1d0d74fb-372f-41ca-9974-1a436bba6ed1, fb867aff-a70c-48ac-aa85-042ca9092f57, 25999b72-0e07-4015-b6e5-1050a04db48f, 2b8c7a11-4c4a-447f-8ae0-e1a3d3a249eb]>
  

но затем в моей рабочей консоли я не вижу обработанных задач

Если я запускаю задачу таким образом — все в порядке

 for i in xrange(10000):        
    my_task_A.delay(i, i)
  

Я пытаюсь оптимизировать (не уверен, есть ли польза) отправку нескольких идентификаторов в одну и ту же задачу вместо того, чтобы выполнять их последовательно.

Чего мне не хватает?

большое спасибо

Обсуждаемые в celery mail group canvas не наследуют маршрутизацию. Предложенный там создателем сельдерея, вы должны называть его таким образом

  target_task.chunks(…)(queue='for_task_A') 
  

Я не пробовал сам, поэтому не могу гарантировать, что это сработает.

Я надеюсь, что это выведет вас на правильный путь.