routing.rst 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553
  1. .. _guide-routing:
  2. ===============
  3. Routing Tasks
  4. ===============
  5. .. warning::
  6. This document refers to functionality only available in brokers
  7. using AMQP. Other brokers may implement some functionality, see their
  8. respective documenation for more information, or contact the :ref:`mailing-list`.
  9. .. contents::
  10. :local:
  11. .. _routing-basics:
  12. Basics
  13. ======
  14. .. _routing-automatic:
  15. Automatic routing
  16. -----------------
  17. The simplest way to do routing is to use the
  18. :setting:`CELERY_CREATE_MISSING_QUEUES` setting (on by default).
  19. With this setting on, a named queue that is not already defined in
  20. :setting:`CELERY_QUEUES` will be created automatically. This makes it easy to
  21. perform simple routing tasks.
  22. Say you have two servers, ``x``, and ``y`` that handles regular tasks,
  23. and one server ``z``, that only handles feed related tasks. You can use this
  24. configuration::
  25. CELERY_ROUTES = {"feed.tasks.import_feed": {"queue": "feeds"}}
  26. With this route enabled import feed tasks will be routed to the
  27. ``"feeds"`` queue, while all other tasks will be routed to the default queue
  28. (named ``"celery"`` for historic reasons).
  29. Now you can start server ``z`` to only process the feeds queue like this::
  30. (z)$ celeryd -Q feeds
  31. You can specify as many queues as you want, so you can make this server
  32. process the default queue as well::
  33. (z)$ celeryd -Q feeds,celery
  34. .. _routing-changing-default-queue:
  35. Changing the name of the default queue
  36. ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  37. You can change the name of the default queue by using the following
  38. configuration:
  39. .. code-block:: python
  40. CELERY_QUEUES = {"default": {"exchange": "default",
  41. "binding_key": "default"}}
  42. CELERY_DEFAULT_QUEUE = "default"
  43. .. _routing-autoqueue-details:
  44. How the queues are defined
  45. ~~~~~~~~~~~~~~~~~~~~~~~~~~
  46. The point with this feature is to hide the complex AMQP protocol for users
  47. with only basic needs. However — you may still be interested in how these queues
  48. are defined.
  49. A queue named ``"video"`` will be created with the following settings:
  50. .. code-block:: python
  51. {"exchange": "video",
  52. "exchange_type": "direct",
  53. "routing_key": "video"}
  54. The non-AMQP backends like ``ghettoq`` does not support exchanges, so they
  55. require the exchange to have the same name as the queue. Using this design
  56. ensures it will work for them as well.
  57. .. _routing-manual:
  58. Manual routing
  59. --------------
  60. Say you have two servers, ``x``, and ``y`` that handles regular tasks,
  61. and one server ``z``, that only handles feed related tasks, you can use this
  62. configuration:
  63. .. code-block:: python
  64. CELERY_DEFAULT_QUEUE = "default"
  65. CELERY_QUEUES = {
  66. "default": {
  67. "binding_key": "task.#",
  68. },
  69. "feed_tasks": {
  70. "binding_key": "feed.#",
  71. },
  72. }
  73. CELERY_DEFAULT_EXCHANGE = "tasks"
  74. CELERY_DEFAULT_EXCHANGE_TYPE = "topic"
  75. CELERY_DEFAULT_ROUTING_KEY = "task.default"
  76. :setting:`CELERY_QUEUES` is a map of queue names and their
  77. exchange/type/binding_key, if you don't set exchange or exchange type, they
  78. will be taken from the :setting:`CELERY_DEFAULT_EXCHANGE` and
  79. :setting:`CELERY_DEFAULT_EXCHANGE_TYPE` settings.
  80. To route a task to the ``feed_tasks`` queue, you can add an entry in the
  81. :setting:`CELERY_ROUTES` setting:
  82. .. code-block:: python
  83. CELERY_ROUTES = {
  84. "feeds.tasks.import_feed": {
  85. "queue": "feed_tasks",
  86. "routing_key": "feed.import",
  87. },
  88. }
  89. You can also override this using the ``routing_key`` argument to
  90. :func:`~celery.execute.apply_async`, or :func:`~celery.execute.send_task`:
  91. >>> from feeds.tasks import import_feed
  92. >>> import_feed.apply_async(args=["http://cnn.com/rss"],
  93. ... queue="feed_tasks",
  94. ... routing_key="feed.import")
  95. To make server ``z`` consume from the feed queue exclusively you can
  96. start it with the ``-Q`` option::
  97. (z)$ celeryd -Q feed_tasks --hostname=z.example.com
  98. Servers ``x`` and ``y`` must be configured to consume from the default queue::
  99. (x)$ celeryd -Q default --hostname=x.example.com
  100. (y)$ celeryd -Q default --hostname=y.example.com
  101. If you want, you can even have your feed processing worker handle regular
  102. tasks as well, maybe in times when there's a lot of work to do::
  103. (z)$ celeryd -Q feed_tasks,default --hostname=z.example.com
  104. If you have another queue but on another exchange you want to add,
  105. just specify a custom exchange and exchange type:
  106. .. code-block:: python
  107. CELERY_QUEUES = {
  108. "feed_tasks": {
  109. "binding_key": "feed.#",
  110. },
  111. "regular_tasks": {
  112. "binding_key": "task.#",
  113. },
  114. "image_tasks": {
  115. "binding_key": "image.compress",
  116. "exchange": "mediatasks",
  117. "exchange_type": "direct",
  118. },
  119. }
  120. If you're confused about these terms, you should read up on AMQP concepts.
  121. .. seealso::
  122. In addition to the :ref:`amqp-primer` below, there's
  123. `Rabbits and Warrens`_, an excellent blog post describing queues and
  124. exchanges. There's also AMQP in 10 minutes*: `Flexible Routing Model`_,
  125. and `Standard Exchange Types`_. For users of RabbitMQ the `RabbitMQ FAQ`_
  126. could be useful as a source of information.
  127. .. _`Rabbits and Warrens`: http://blogs.digitar.com/jjww/2009/01/rabbits-and-warrens/
  128. .. _`Flexible Routing Model`: http://bit.ly/95XFO1
  129. .. _`Standard Exchange Types`: http://bit.ly/EEWca
  130. .. _`RabbitMQ FAQ`: http://www.rabbitmq.com/faq.html
  131. .. _amqp-primer:
  132. AMQP Primer
  133. ===========
  134. Messages
  135. --------
  136. A message consists of headers and a body. Celery uses headers to store
  137. the content type of the message and its content encoding. In Celery the
  138. content type is usually the serialization format used to serialize the
  139. message, and the body contains the name of the task to execute, the
  140. task id (UUID), the arguments to execute it with and some additional
  141. metadata - like the number of retries and its ETA (if any).
  142. This is an example task message represented as a Python dictionary:
  143. .. code-block:: python
  144. {"task": "myapp.tasks.add",
  145. "id": "54086c5e-6193-4575-8308-dbab76798756",
  146. "args": [4, 4],
  147. "kwargs": {}}
  148. .. _amqp-producers-consumers-brokers:
  149. Producers, consumers and brokers
  150. --------------------------------
  151. The client sending messages is typically called a *publisher*, or
  152. a *producer*, while the entity receiving messages is called
  153. a *consumer*.
  154. The *broker* is the message server, routing messages from producers
  155. to consumers.
  156. You are likely to see these terms used a lot in AMQP related material.
  157. .. _amqp-exchanges-queues-keys:
  158. Exchanges, queues and routing keys.
  159. -----------------------------------
  160. 1. Messages are sent to exchanges.
  161. 2. An exchange routes messages to one or more queues. Several exchange types
  162. exists, providing different ways to do routing.
  163. 3. The message waits in the queue until someone consumes from it.
  164. 4. The message is deleted from the queue when it has been acknowledged.
  165. The steps required to send and receive messages are:
  166. 1. Create an exchange
  167. 2. Create a queue
  168. 3. Bind the queue to the exchange.
  169. Celery automatically creates the entities necessary for the queues in
  170. :setting:`CELERY_QUEUES` to work (except if the queue's ``auto_declare``
  171. setting is set to :const:`False`).
  172. Here's an example queue configuration with three queues;
  173. One for video, one for images and finally, one default queue for everything else:
  174. .. code-block:: python
  175. CELERY_QUEUES = {
  176. "default": {
  177. "exchange": "default",
  178. "binding_key": "default"},
  179. "videos": {
  180. "exchange": "media",
  181. "binding_key": "media.video",
  182. },
  183. "images": {
  184. "exchange": "media",
  185. "binding_key": "media.image",
  186. }
  187. }
  188. CELERY_DEFAULT_QUEUE = "default"
  189. CELERY_DEFAULT_EXCHANGE_TYPE = "direct"
  190. CELERY_DEFAULT_ROUTING_KEY = "default"
  191. .. note::
  192. In Celery the ``routing_key`` is the key used to send the message,
  193. while ``binding_key`` is the key the queue is bound with. In the AMQP API
  194. they are both referred to as the routing key.
  195. .. _amqp-exchange-types:
  196. Exchange types
  197. --------------
  198. The exchange type defines how the messages are routed through the exchange.
  199. The exchange types defined in the standard are ``direct``, ``topic``,
  200. ``fanout`` and ``headers``. Also non-standard exchange types are available
  201. as plugins to RabbitMQ, like the `last-value-cache plug-in`_ by Michael
  202. Bridgen.
  203. .. _`last-value-cache plug-in`:
  204. http://github.com/squaremo/rabbitmq-lvc-plugin
  205. .. _amqp-exchange-type-direct:
  206. Direct exchanges
  207. ~~~~~~~~~~~~~~~~
  208. Direct exchanges match by exact routing keys, so a queue bound with
  209. the routing key ``video`` only receives messages with the same routing key.
  210. .. _amqp-exchange-type-topic:
  211. Topic exchanges
  212. ~~~~~~~~~~~~~~~
  213. Topic exchanges matches routing keys using dot-separated words, and can
  214. include wildcard characters: ``*`` matches a single word, ``#`` matches
  215. zero or more words.
  216. With routing keys like ``usa.news``, ``usa.weather``, ``norway.news`` and
  217. ``norway.weather``, bindings could be ``*.news`` (all news), ``usa.#`` (all
  218. items in the USA) or ``usa.weather`` (all USA weather items).
  219. .. _amqp-api:
  220. Related API commands
  221. --------------------
  222. .. method:: exchange.declare(exchange_name, type, passive,
  223. durable, auto_delete, internal)
  224. Declares an exchange by name.
  225. :keyword passive: Passive means the exchange won't be created, but you
  226. can use this to check if the exchange already exists.
  227. :keyword durable: Durable exchanges are persistent. That is - they survive
  228. a broker restart.
  229. :keyword auto_delete: This means the queue will be deleted by the broker
  230. when there are no more queues using it.
  231. .. method:: queue.declare(queue_name, passive, durable, exclusive, auto_delete)
  232. Declares a queue by name.
  233. Exclusive queues can only be consumed from by the current connection.
  234. Exclusive also implies ``auto_delete``.
  235. .. method:: queue.bind(queue_name, exchange_name, routing_key)
  236. Binds a queue to an exchange with a routing key.
  237. Unbound queues will not receive messages, so this is necessary.
  238. .. method:: queue.delete(name, if_unused=False, if_empty=False)
  239. Deletes a queue and its binding.
  240. .. method:: exchange.delete(name, if_unused=False)
  241. Deletes an exchange.
  242. .. note::
  243. Declaring does not necessarily mean "create". When you declare you
  244. *assert* that the entity exists and that it's operable. There is no
  245. rule as to whom should initially create the exchange/queue/binding,
  246. whether consumer or producer. Usually the first one to need it will
  247. be the one to create it.
  248. .. _amqp-api-hands-on:
  249. Hands-on with the API
  250. ---------------------
  251. Celery comes with a tool called ``camqadm`` (short for celery AMQP admin).
  252. It's used for simple admnistration tasks like creating/deleting queues and
  253. exchanges, purging queues and sending messages. In short it's for simple
  254. command-line access to the AMQP API.
  255. You can write commands directly in the arguments to ``camqadm``, or just start
  256. with no arguments to start it in shell-mode::
  257. $ camqadm
  258. -> connecting to amqp://guest@localhost:5672/.
  259. -> connected.
  260. 1>
  261. Here ``1>`` is the prompt. The number is counting the number of commands you
  262. have executed. Type ``help`` for a list of commands. It also has
  263. autocompletion, so you can start typing a command and then hit the
  264. ``tab`` key to show a list of possible matches.
  265. Now let's create a queue we can send messages to::
  266. 1> exchange.declare testexchange direct
  267. ok.
  268. 2> queue.declare testqueue
  269. ok. queue:testqueue messages:0 consumers:0.
  270. 3> queue.bind testqueue testexchange testkey
  271. ok.
  272. This created the direct exchange ``testexchange``, and a queue
  273. named ``testqueue``. The queue is bound to the exchange using
  274. the routing key ``testkey``.
  275. From now on all messages sent to the exchange ``testexchange`` with routing
  276. key ``testkey`` will be moved to this queue. We can send a message by
  277. using the ``basic.publish`` command::
  278. 4> basic.publish "This is a message!" testexchange testkey
  279. ok.
  280. Now that the message is sent we can retrieve it again. We use the
  281. ``basic.get`` command here, which pops a single message off the queue,
  282. this command is not recommended for production as it implies polling, any
  283. real application would declare consumers instead.
  284. Pop a message off the queue::
  285. 5> basic.get testqueue
  286. {'body': 'This is a message!',
  287. 'delivery_info': {'delivery_tag': 1,
  288. 'exchange': u'testexchange',
  289. 'message_count': 0,
  290. 'redelivered': False,
  291. 'routing_key': u'testkey'},
  292. 'properties': {}}
  293. AMQP uses acknowledgment to signify that a message has been received
  294. and processed successfully. The message is sent to the next receiver
  295. if it has not been acknowledged before the client connection is closed.
  296. Note the delivery tag listed in the structure above; Within a connection channel,
  297. every received message has a unique delivery tag,
  298. This tag is used to acknowledge the message. Also note that
  299. delivery tags are not unique across connections, so in another client
  300. the delivery tag ``1`` might point to a different message than in this channel.
  301. You can acknowledge the message we received using ``basic.ack``::
  302. 6> basic.ack 1
  303. ok.
  304. To clean up after our test session we should delete the entities we created::
  305. 7> queue.delete testqueue
  306. ok. 0 messages deleted.
  307. 8> exchange.delete testexchange
  308. ok.
  309. .. _routing-tasks:
  310. Routing Tasks
  311. =============
  312. .. _routing-defining-queues:
  313. Defining queues
  314. ---------------
  315. In Celery the queues are defined by the :setting:`CELERY_QUEUES` setting.
  316. Here's an example queue configuration with three queues;
  317. One for video, one for images and finally, one default queue for everything else:
  318. .. code-block:: python
  319. CELERY_QUEUES = {
  320. "default": {
  321. "exchange": "default",
  322. "binding_key": "default"},
  323. "videos": {
  324. "exchange": "media",
  325. "exchange_type": "topic",
  326. "binding_key": "media.video",
  327. },
  328. "images": {
  329. "exchange": "media",
  330. "exchange_type": "topic",
  331. "binding_key": "media.image",
  332. }
  333. }
  334. CELERY_DEFAULT_QUEUE = "default"
  335. CELERY_DEFAULT_EXCHANGE = "default"
  336. CELERY_DEFAULT_EXCHANGE_TYPE = "direct"
  337. CELERY_DEFAULT_ROUTING_KEY = "default"
  338. Here, the :setting:`CELERY_DEFAULT_QUEUE` will be used to route tasks that
  339. doesn't have an explicit route.
  340. The default exchange, exchange type and routing key will be used as the
  341. default routing values for tasks, and as the default values for entries
  342. in :setting:`CELERY_QUEUES`.
  343. .. _routing-task-destination:
  344. Specifying task destination
  345. ---------------------------
  346. The destination for a task is decided by the following (in order):
  347. 1. The :ref:`routers` defined in :setting:`CELERY_ROUTES`.
  348. 2. The routing arguments to :func:`~celery.execute.apply_async`.
  349. 3. Routing related attributes defined on the :class:`~celery.task.base.Task`
  350. itself.
  351. It is considered best practice to not hard-code these settings, but rather
  352. leave that as configuration options by using :ref:`routers`;
  353. This is the most flexible approach, but sensible defaults can still be set
  354. as task attributes.
  355. .. _routers:
  356. Routers
  357. -------
  358. A router is a class that decides the routing options for a task.
  359. All you need to define a new router is to create a class with a
  360. ``route_for_task`` method:
  361. .. code-block:: python
  362. class MyRouter(object):
  363. def route_for_task(self, task, args=None, kwargs=None):
  364. if task == "myapp.tasks.compress_video":
  365. return {"exchange": "video",
  366. "exchange_type": "topic",
  367. "routing_key": "video.compress"}
  368. return None
  369. If you return the ``queue`` key, it will expand with the defined settings of
  370. that queue in :setting:`CELERY_QUEUES`::
  371. {"queue": "video", "routing_key": "video.compress"}
  372. becomes -->
  373. {"queue": "video",
  374. "exchange": "video",
  375. "exchange_type": "topic",
  376. "routing_key": "video.compress"}
  377. You install router classes by adding it to the :setting:`CELERY_ROUTES` setting::
  378. CELERY_ROUTES = (MyRouter, )
  379. Router classes can also be added by name::
  380. CELERY_ROUTES = ("myapp.routers.MyRouter", )
  381. For simple task name -> route mappings like the router example above, you can simply
  382. drop a dict into :setting:`CELERY_ROUTES` to get the same result::
  383. CELERY_ROUTES = ({"myapp.tasks.compress_video": {
  384. "queue": "video",
  385. "routing_key": "video.compress"}}, )
  386. The routers will then be traversed in order, it will stop at the first router
  387. returning a value and use that as the final route for the task.