routing.rst 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715
  1. .. _guide-routing:
  2. ===============
  3. Routing Tasks
  4. ===============
  5. .. note::
  6. Alternate routing concepts like topic and fanout is not
  7. available for all transports, please consult the
  8. :ref:`transport comparison table <kombu:transport-comparison>`.
  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:`task_create_missing_queues` setting (on by default).
  19. With this setting on, a named queue that's not already defined in
  20. :setting:`task_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. task_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 historical reasons).
  29. Alternatively, you can use glob pattern matching, or even regular expressions,
  30. to match all tasks in the ``feed.tasks`` name-space:
  31. .. code-block:: python
  32. app.conf.task_routes = {'feed.tasks.*': {'queue': 'feeds'}}
  33. If the order of matching patterns is important you should
  34. specify the router in *items* format instead:
  35. .. code-block:: python
  36. task_routes = ([
  37. ('feed.tasks.*', {'queue': 'feeds'}),
  38. ('web.tasks.*', {'queue': 'web'}),
  39. (re.compile(r'(video|image)\.tasks\..*'), {'queue': 'media'}),
  40. ],)
  41. .. note::
  42. The :setting:`task_routes` setting can either be a dictionary, or a
  43. list of router objects, so in this case we need to specify the setting
  44. as a tuple containing a list.
  45. After installing the router, you can start server `z` to only process the feeds
  46. queue like this:
  47. .. code-block:: console
  48. user@z:/$ celery -A proj worker -Q feeds
  49. You can specify as many queues as you want, so you can make this server
  50. process the default queue as well:
  51. .. code-block:: console
  52. user@z:/$ celery -A proj worker -Q feeds,celery
  53. .. _routing-changing-default-queue:
  54. Changing the name of the default queue
  55. ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  56. You can change the name of the default queue by using the following
  57. configuration:
  58. .. code-block:: python
  59. app.conf.task_default_queue = 'default'
  60. .. _routing-autoqueue-details:
  61. How the queues are defined
  62. ~~~~~~~~~~~~~~~~~~~~~~~~~~
  63. The point with this feature is to hide the complex AMQP protocol for users
  64. with only basic needs. However -- you may still be interested in how these queues
  65. are declared.
  66. A queue named `"video"` will be created with the following settings:
  67. .. code-block:: javascript
  68. {'exchange': 'video',
  69. 'exchange_type': 'direct',
  70. 'routing_key': 'video'}
  71. The non-AMQP backends like `Redis` or `SQS` don't support exchanges,
  72. so they require the exchange to have the same name as the queue. Using this
  73. design ensures it will work for them as well.
  74. .. _routing-manual:
  75. Manual routing
  76. --------------
  77. Say you have two servers, `x`, and `y` that handles regular tasks,
  78. and one server `z`, that only handles feed related tasks, you can use this
  79. configuration:
  80. .. code-block:: python
  81. from kombu import Queue
  82. app.conf.task_default_queue = 'default'
  83. app.conf.task_queues = (
  84. Queue('default', routing_key='task.#'),
  85. Queue('feed_tasks', routing_key='feed.#'),
  86. )
  87. task_default_exchange = 'tasks'
  88. task_default_exchange_type = 'topic'
  89. task_default_routing_key = 'task.default'
  90. :setting:`task_queues` is a list of :class:`~kombu.entitity.Queue`
  91. instances.
  92. If you don't set the exchange or exchange type values for a key, these
  93. will be taken from the :setting:`task_default_exchange` and
  94. :setting:`task_default_exchange_type` settings.
  95. To route a task to the `feed_tasks` queue, you can add an entry in the
  96. :setting:`task_routes` setting:
  97. .. code-block:: python
  98. task_routes = {
  99. 'feeds.tasks.import_feed': {
  100. 'queue': 'feed_tasks',
  101. 'routing_key': 'feed.import',
  102. },
  103. }
  104. You can also override this using the `routing_key` argument to
  105. :meth:`Task.apply_async`, or :func:`~celery.execute.send_task`:
  106. >>> from feeds.tasks import import_feed
  107. >>> import_feed.apply_async(args=['http://cnn.com/rss'],
  108. ... queue='feed_tasks',
  109. ... routing_key='feed.import')
  110. To make server `z` consume from the feed queue exclusively you can
  111. start it with the :option:`celery worker -Q` option:
  112. .. code-block:: console
  113. user@z:/$ celery -A proj worker -Q feed_tasks --hostname=z@%h
  114. Servers `x` and `y` must be configured to consume from the default queue:
  115. .. code-block:: console
  116. user@x:/$ celery -A proj worker -Q default --hostname=x@%h
  117. user@y:/$ celery -A proj worker -Q default --hostname=y@%h
  118. If you want, you can even have your feed processing worker handle regular
  119. tasks as well, maybe in times when there's a lot of work to do:
  120. .. code-block:: console
  121. user@z:/$ celery -A proj worker -Q feed_tasks,default --hostname=z@%h
  122. If you have another queue but on another exchange you want to add,
  123. just specify a custom exchange and exchange type:
  124. .. code-block:: python
  125. from kombu import Exchange, Queue
  126. app.conf.task_queues = (
  127. Queue('feed_tasks', routing_key='feed.#'),
  128. Queue('regular_tasks', routing_key='task.#'),
  129. Queue('image_tasks', exchange=Exchange('mediatasks', type='direct'),
  130. routing_key='image.compress'),
  131. )
  132. If you're confused about these terms, you should read up on AMQP.
  133. .. seealso::
  134. In addition to the :ref:`amqp-primer` below, there's
  135. `Rabbits and Warrens`_, an excellent blog post describing queues and
  136. exchanges. There's also The `CloudAMQP tutorial`,
  137. For users of RabbitMQ the `RabbitMQ FAQ`_
  138. could be useful as a source of information.
  139. .. _`Rabbits and Warrens`: http://blogs.digitar.com/jjww/2009/01/rabbits-and-warrens/
  140. .. _`CloudAMQP tutorial`: amqp in 10 minutes part 3
  141. https://www.cloudamqp.com/blog/2015-09-03-part4-rabbitmq-for-beginners-exchanges-routing-keys-bindings.html
  142. .. _`RabbitMQ FAQ`: https://www.rabbitmq.com/faq.html
  143. .. _routing-special_options:
  144. Special Routing Options
  145. =======================
  146. .. _routing-options-rabbitmq-priorities:
  147. RabbitMQ Message Priorities
  148. ---------------------------
  149. :supported transports: RabbitMQ
  150. .. versionadded:: 4.0
  151. Queues can be configured to support priorities by setting the
  152. ``x-max-priority`` argument:
  153. .. code-block:: python
  154. from kombu import Exchange, Queue
  155. app.conf.task_queues = [
  156. Queue('tasks', Exchange('tasks'), routing_key='tasks',
  157. queue_arguments={'x-max-priority': 10},
  158. ]
  159. A default value for all queues can be set using the
  160. :setting:`task_queue_max_priority` setting:
  161. .. code-block:: python
  162. app.conf.task_queue_max_priority = 10
  163. .. _amqp-primer:
  164. AMQP Primer
  165. ===========
  166. Messages
  167. --------
  168. A message consists of headers and a body. Celery uses headers to store
  169. the content type of the message and its content encoding. The
  170. content type is usually the serialization format used to serialize the
  171. message. The body contains the name of the task to execute, the
  172. task id (UUID), the arguments to apply it with and some additional
  173. meta-data -- like the number of retries or an ETA.
  174. This is an example task message represented as a Python dictionary:
  175. .. code-block:: javascript
  176. {'task': 'myapp.tasks.add',
  177. 'id': '54086c5e-6193-4575-8308-dbab76798756',
  178. 'args': [4, 4],
  179. 'kwargs': {}}
  180. .. _amqp-producers-consumers-brokers:
  181. Producers, consumers, and brokers
  182. ---------------------------------
  183. The client sending messages is typically called a *publisher*, or
  184. a *producer*, while the entity receiving messages is called
  185. a *consumer*.
  186. The *broker* is the message server, routing messages from producers
  187. to consumers.
  188. You're likely to see these terms used a lot in AMQP related material.
  189. .. _amqp-exchanges-queues-keys:
  190. Exchanges, queues, and routing keys
  191. -----------------------------------
  192. 1. Messages are sent to exchanges.
  193. 2. An exchange routes messages to one or more queues. Several exchange types
  194. exists, providing different ways to do routing, or implementing
  195. different messaging scenarios.
  196. 3. The message waits in the queue until someone consumes it.
  197. 4. The message is deleted from the queue when it has been acknowledged.
  198. The steps required to send and receive messages are:
  199. 1. Create an exchange
  200. 2. Create a queue
  201. 3. Bind the queue to the exchange.
  202. Celery automatically creates the entities necessary for the queues in
  203. :setting:`task_queues` to work (except if the queue's `auto_declare`
  204. setting is set to :const:`False`).
  205. Here's an example queue configuration with three queues;
  206. One for video, one for images, and one default queue for everything else:
  207. .. code-block:: python
  208. from kombu import Exchange, Queue
  209. app.conf.task_queues = (
  210. Queue('default', Exchange('default'), routing_key='default'),
  211. Queue('videos', Exchange('media'), routing_key='media.video'),
  212. Queue('images', Exchange('media'), routing_key='media.image'),
  213. )
  214. app.conf.task_default_queue = 'default'
  215. app.conf.task_default_exchange_type = 'direct'
  216. app.conf.task_default_routing_key = 'default'
  217. .. _amqp-exchange-types:
  218. Exchange types
  219. --------------
  220. The exchange type defines how the messages are routed through the exchange.
  221. The exchange types defined in the standard are `direct`, `topic`,
  222. `fanout` and `headers`. Also non-standard exchange types are available
  223. as plug-ins to RabbitMQ, like the `last-value-cache plug-in`_ by Michael
  224. Bridgen.
  225. .. _`last-value-cache plug-in`:
  226. https://github.com/squaremo/rabbitmq-lvc-plugin
  227. .. _amqp-exchange-type-direct:
  228. Direct exchanges
  229. ~~~~~~~~~~~~~~~~
  230. Direct exchanges match by exact routing keys, so a queue bound by
  231. the routing key `video` only receives messages with that routing key.
  232. .. _amqp-exchange-type-topic:
  233. Topic exchanges
  234. ~~~~~~~~~~~~~~~
  235. Topic exchanges matches routing keys using dot-separated words, and the
  236. wild-card characters: ``*`` (matches a single word), and ``#`` (matches
  237. zero or more words).
  238. With routing keys like ``usa.news``, ``usa.weather``, ``norway.news``, and
  239. ``norway.weather``, bindings could be ``*.news`` (all news), ``usa.#`` (all
  240. items in the USA), or ``usa.weather`` (all USA weather items).
  241. .. _amqp-api:
  242. Related API commands
  243. --------------------
  244. .. method:: exchange.declare(exchange_name, type, passive,
  245. durable, auto_delete, internal)
  246. Declares an exchange by name.
  247. See :meth:`amqp:Channel.exchange_declare <amqp.channel.Channel.exchange_declare>`.
  248. :keyword passive: Passive means the exchange won't be created, but you
  249. can use this to check if the exchange already exists.
  250. :keyword durable: Durable exchanges are persistent (i.e., they survive
  251. a broker restart).
  252. :keyword auto_delete: This means the exchange will be deleted by the broker
  253. when there are no more queues using it.
  254. .. method:: queue.declare(queue_name, passive, durable, exclusive, auto_delete)
  255. Declares a queue by name.
  256. See :meth:`amqp:Channel.queue_declare <amqp.channel.Channel.queue_declare>`
  257. Exclusive queues can only be consumed from by the current connection.
  258. Exclusive also implies `auto_delete`.
  259. .. method:: queue.bind(queue_name, exchange_name, routing_key)
  260. Binds a queue to an exchange with a routing key.
  261. Unbound queues won't receive messages, so this is necessary.
  262. See :meth:`amqp:Channel.queue_bind <amqp.channel.Channel.queue_bind>`
  263. .. method:: queue.delete(name, if_unused=False, if_empty=False)
  264. Deletes a queue and its binding.
  265. See :meth:`amqp:Channel.queue_delete <amqp.channel.Channel.queue_delete>`
  266. .. method:: exchange.delete(name, if_unused=False)
  267. Deletes an exchange.
  268. See :meth:`amqp:Channel.exchange_delete <amqp.channel.Channel.exchange_delete>`
  269. .. note::
  270. Declaring doesn't necessarily mean "create". When you declare you
  271. *assert* that the entity exists and that it's operable. There's no
  272. rule as to whom should initially create the exchange/queue/binding,
  273. whether consumer or producer. Usually the first one to need it will
  274. be the one to create it.
  275. .. _amqp-api-hands-on:
  276. Hands-on with the API
  277. ---------------------
  278. Celery comes with a tool called :program:`celery amqp`
  279. that's used for command line access to the AMQP API, enabling access to
  280. administration tasks like creating/deleting queues and exchanges, purging
  281. queues or sending messages. It can also be used for non-AMQP brokers,
  282. but different implementation may not implement all commands.
  283. You can write commands directly in the arguments to :program:`celery amqp`,
  284. or just start with no arguments to start it in shell-mode:
  285. .. code-block:: console
  286. $ celery -A proj amqp
  287. -> connecting to amqp://guest@localhost:5672/.
  288. -> connected.
  289. 1>
  290. Here ``1>`` is the prompt. The number 1, is the number of commands you
  291. have executed so far. Type ``help`` for a list of commands available.
  292. It also supports auto-completion, so you can start typing a command and then
  293. hit the `tab` key to show a list of possible matches.
  294. Let's create a queue you can send messages to:
  295. .. code-block:: console
  296. $ celery -A proj amqp
  297. 1> exchange.declare testexchange direct
  298. ok.
  299. 2> queue.declare testqueue
  300. ok. queue:testqueue messages:0 consumers:0.
  301. 3> queue.bind testqueue testexchange testkey
  302. ok.
  303. This created the direct exchange ``testexchange``, and a queue
  304. named ``testqueue``. The queue is bound to the exchange using
  305. the routing key ``testkey``.
  306. From now on all messages sent to the exchange ``testexchange`` with routing
  307. key ``testkey`` will be moved to this queue. You can send a message by
  308. using the ``basic.publish`` command:
  309. .. code-block:: console
  310. 4> basic.publish 'This is a message!' testexchange testkey
  311. ok.
  312. Now that the message is sent you can retrieve it again. You can use the
  313. ``basic.get``` command here, that polls for new messages on the queue
  314. in a synchronous manner
  315. (this is OK for maintenance tasks, but for services you want to use
  316. ``basic.consume`` instead)
  317. Pop a message off the queue:
  318. .. code-block:: console
  319. 5> basic.get testqueue
  320. {'body': 'This is a message!',
  321. 'delivery_info': {'delivery_tag': 1,
  322. 'exchange': u'testexchange',
  323. 'message_count': 0,
  324. 'redelivered': False,
  325. 'routing_key': u'testkey'},
  326. 'properties': {}}
  327. AMQP uses acknowledgment to signify that a message has been received
  328. and processed successfully. If the message hasn't been acknowledged
  329. and consumer channel is closed, the message will be delivered to
  330. another consumer.
  331. Note the delivery tag listed in the structure above; Within a connection
  332. channel, every received message has a unique delivery tag,
  333. This tag is used to acknowledge the message. Also note that
  334. delivery tags aren't unique across connections, so in another client
  335. the delivery tag `1` might point to a different message than in this channel.
  336. You can acknowledge the message you received using ``basic.ack``:
  337. .. code-block:: console
  338. 6> basic.ack 1
  339. ok.
  340. To clean up after our test session you should delete the entities you created:
  341. .. code-block:: console
  342. 7> queue.delete testqueue
  343. ok. 0 messages deleted.
  344. 8> exchange.delete testexchange
  345. ok.
  346. .. _routing-tasks:
  347. Routing Tasks
  348. =============
  349. .. _routing-defining-queues:
  350. Defining queues
  351. ---------------
  352. In Celery available queues are defined by the :setting:`task_queues` setting.
  353. Here's an example queue configuration with three queues;
  354. One for video, one for images, and one default queue for everything else:
  355. .. code-block:: python
  356. default_exchange = Exchange('default', type='direct')
  357. media_exchange = Exchange('media', type='direct')
  358. app.conf.task_queues = (
  359. Queue('default', default_exchange, routing_key='default'),
  360. Queue('videos', media_exchange, routing_key='media.video'),
  361. Queue('images', media_exchange, routing_key='media.image')
  362. )
  363. app.conf.task_default_queue = 'default'
  364. app.conf.task_default_exchange = 'default'
  365. app.conf.task_default_routing_key = 'default'
  366. Here, the :setting:`task_default_queue` will be used to route tasks that
  367. doesn't have an explicit route.
  368. The default exchange, exchange type, and routing key will be used as the
  369. default routing values for tasks, and as the default values for entries
  370. in :setting:`task_queues`.
  371. Multiple bindings to a single queue are also supported. Here's an example
  372. of two routing keys that are both bound to the same queue:
  373. .. code-block:: python
  374. from kombu import Exchange, Queue, binding
  375. media_exchange = Exchange('media', type='direct')
  376. CELERY_QUEUES = (
  377. Queue('media', [
  378. binding(media_exchange, routing_key='media.video'),
  379. binding(media_exchange, routing_key='media.image'),
  380. ]),
  381. )
  382. .. _routing-task-destination:
  383. Specifying task destination
  384. ---------------------------
  385. The destination for a task is decided by the following (in order):
  386. 1. The :ref:`routers` defined in :setting:`task_routes`.
  387. 2. The routing arguments to :func:`Task.apply_async`.
  388. 3. Routing related attributes defined on the :class:`~celery.task.base.Task`
  389. itself.
  390. It's considered best practice to not hard-code these settings, but rather
  391. leave that as configuration options by using :ref:`routers`;
  392. This is the most flexible approach, but sensible defaults can still be set
  393. as task attributes.
  394. .. _routers:
  395. Routers
  396. -------
  397. A router is a function that decides the routing options for a task.
  398. All you need to define a new router is to define a function with
  399. the signature ``(name, args, kwargs, options, task=None, **kw)``:
  400. .. code-block:: python
  401. def route_task(name, args, kwargs, options, task=None, **kw):
  402. if name == 'myapp.tasks.compress_video':
  403. return {'exchange': 'video',
  404. 'exchange_type': 'topic',
  405. 'routing_key': 'video.compress'}
  406. If you return the ``queue`` key, it'll expand with the defined settings of
  407. that queue in :setting:`task_queues`:
  408. .. code-block:: javascript
  409. {'queue': 'video', 'routing_key': 'video.compress'}
  410. becomes -->
  411. .. code-block:: javascript
  412. {'queue': 'video',
  413. 'exchange': 'video',
  414. 'exchange_type': 'topic',
  415. 'routing_key': 'video.compress'}
  416. You install router classes by adding them to the :setting:`task_routes`
  417. setting:
  418. .. code-block:: python
  419. task_routes = (route_task,)
  420. Router functions can also be added by name:
  421. .. code-block:: python
  422. task_routes = ('myapp.routers.route_task',)
  423. For simple task name -> route mappings like the router example above,
  424. you can simply drop a dict into :setting:`task_routes` to get the
  425. same behavior:
  426. .. code-block:: python
  427. task_routes = {
  428. 'myapp.tasks.compress_video': {
  429. 'queue': 'video',
  430. 'routing_key': 'video.compress',
  431. },
  432. }
  433. The routers will then be traversed in order, it will stop at the first router
  434. returning a true value, and use that as the final route for the task.
  435. You can also have multiple routers defined in a sequence:
  436. .. code-block:: python
  437. task_routes = [
  438. route_task,
  439. {
  440. 'myapp.tasks.compress_video': {
  441. 'queue': 'video',
  442. 'routing_key': 'video.compress',
  443. },
  444. ]
  445. The routers will then be visited in turn, and the first to return
  446. a value will be chosen.
  447. Broadcast
  448. ---------
  449. Celery can also support broadcast routing.
  450. Here is an example exchange ``broadcast_tasks`` that delivers
  451. copies of tasks to all workers connected to it:
  452. .. code-block:: python
  453. from kombu.common import Broadcast
  454. app.conf.task_queues = (Broadcast('broadcast_tasks'),)
  455. app.conf.task_routes = {
  456. 'tasks.reload_cache': {
  457. 'queue': 'broadcast_tasks',
  458. 'exchange': 'broadcast_tasks'
  459. }
  460. }
  461. Now the ``tasks.reload_cache`` task will be sent to every
  462. worker consuming from this queue.
  463. Here is another example of broadcast routing, this time with
  464. a :program:`celery beat` schedule:
  465. .. code-block:: python
  466. from kombu.common import Broadcast
  467. from celery.schedules import crontab
  468. app.conf.task_queues = (Broadcast('broadcast_tasks'),)
  469. app.conf.beat_schedule = {
  470. 'test-task': {
  471. 'task': 'tasks.reload_cache',
  472. 'schedule': crontab(minute=0, hour='*/3'),
  473. 'options': {'exchange': 'broadcast_tasks'}
  474. },
  475. }
  476. .. admonition:: Broadcast & Results
  477. Note that Celery result doesn't define what happens if two
  478. tasks have the same task_id. If the same task is distributed to more
  479. than one worker, then the state history may not be preserved.
  480. It's a good idea to set the ``task.ignore_result`` attribute in
  481. this case.