req.py 840 B

12345678910111213141516171819202122232425262728293031323334353637
  1. from celery import current_app, task, uuid
  2. from celery.five import Queue, range
  3. from celery.worker.consumer import Consumer
  4. from celery.worker.job import Request
  5. from celery.concurrency.solo import TaskPool
  6. from celery.app.amqp import TASK_BARE
  7. from time import time
  8. from librabbitmq import Message
  9. import socket
  10. import sys
  11. @task(accept_magic_kwargs=False)
  12. def T():
  13. pass
  14. tid = uuid()
  15. P = TaskPool()
  16. hostname = socket.gethostname()
  17. task = {'task': T.name, 'args': (), 'kwargs': {}, 'id': tid, 'flags': 0}
  18. app = current_app._get_current_object()
  19. def on_task(req):
  20. req.execute_using_pool(P)
  21. def on_ack(*a): pass
  22. m = Message(None, {}, {}, task)
  23. x = Consumer(on_task, hostname=hostname, app=app)
  24. x.update_strategies()
  25. name = T.name
  26. ts = time()
  27. for i in range(100000):
  28. x.strategies[name](m, m.body, on_ack)
  29. print(time() - ts)