mirror of https://github.com/celery/kombu.git
126 lines
3.5 KiB
ReStructuredText
126 lines
3.5 KiB
ReStructuredText
.. _guide-producers:
|
|
|
|
===========
|
|
Producers
|
|
===========
|
|
|
|
.. _producer-basics:
|
|
|
|
Basics
|
|
======
|
|
|
|
You can create a producer using a :class:`~kombu.Connection`:
|
|
|
|
.. code-block:: pycon
|
|
|
|
>>> producer = connection.Producer()
|
|
|
|
You can also instantiate :class:`~kombu.Producer` directly,
|
|
it takes a channel or a connection as an argument:
|
|
|
|
.. code-block:: pycon
|
|
|
|
>>> with Connection('amqp://') as conn:
|
|
... with conn.channel() as channel:
|
|
... producer = Producer(channel)
|
|
|
|
Having a producer instance you can publish messages:
|
|
|
|
.. code-block:: pycon
|
|
|
|
>>> from kombu import Exchange
|
|
|
|
>>> exchange = Exchange('name', type='direct')
|
|
|
|
>>> producer.publish(
|
|
... {'hello': 'world'}, # message to send
|
|
... exchange=exchange, # destination exchange
|
|
... routing_key='rk', # destination routing key,
|
|
... declare=[exchange], # make sure exchange is declared,
|
|
... )
|
|
|
|
|
|
Mostly you will be getting a connection from a connection pool,
|
|
and this connection can be stale, or you could lose the connection
|
|
in the middle of sending the message. Using retries is a good
|
|
way to handle these intermittent failures:
|
|
|
|
.. code-block:: pycon
|
|
|
|
>>> producer.publish({'hello': 'world', ..., retry=True})
|
|
|
|
In addition a retry policy can be specified, which is a dictionary
|
|
of parameters supported by the :func:`~kombu.utils.functional.retry_over_time`
|
|
function
|
|
|
|
.. code-block:: pycon
|
|
|
|
>>> producer.publish(
|
|
... {'hello': 'world'}, ...,
|
|
... retry=True,
|
|
... retry_policy={
|
|
... 'interval_start': 0, # First retry immediately,
|
|
... 'interval_step': 2, # then increase by 2s for every retry.
|
|
... 'interval_max': 30, # but don't exceed 30s between retries.
|
|
... 'max_retries': 30, # give up after 30 tries.
|
|
... },
|
|
... )
|
|
|
|
The ``declare`` argument lets you pass a list of entities that must be
|
|
declared before sending the message. This is especially important
|
|
when using the ``retry`` flag, since the broker may actually restart
|
|
during a retry in which case non-durable entities are removed.
|
|
|
|
Say you are writing a task queue, and the workers may have not started yet
|
|
so the queues aren't declared. In this case you need to define both the
|
|
exchange, and the declare the queue so that the message is delivered to
|
|
the queue while the workers are offline:
|
|
|
|
.. code-block:: pycon
|
|
|
|
>>> from kombu import Exchange, Queue
|
|
>>> task_queue = Queue('tasks', Exchange('tasks'), routing_key='tasks')
|
|
|
|
>>> producer.publish(
|
|
... {'hello': 'world'}, ...,
|
|
... retry=True,
|
|
... exchange=task_queue.exchange,
|
|
... routing_key=task_queue.routing_key,
|
|
... declare=[task_queue], # declares exchange, queue and binds.
|
|
... )
|
|
|
|
Bypassing routing by using the anon-exchange
|
|
--------------------------------------------
|
|
|
|
You may deliver to a queue directly, bypassing the brokers routing
|
|
mechanisms, by using the "anon-exchange": set the exchange parameter to the
|
|
empty string, and set the routing key to be the name of the queue:
|
|
|
|
.. code-block:: pycon
|
|
|
|
>>> producer.publish(
|
|
... {'hello': 'world'},
|
|
... exchange='',
|
|
... routing_key=task_queue.name,
|
|
... )
|
|
|
|
Serialization
|
|
=============
|
|
|
|
Json is the default serializer when a non-string object is passed
|
|
to publish, but you can also specify a different serializer:
|
|
|
|
.. code-block:: pycon
|
|
|
|
>>> producer.publish({'hello': 'world'}, serializer='pickle')
|
|
|
|
See :ref:`guide-serialization` for more information.
|
|
|
|
|
|
Reference
|
|
=========
|
|
|
|
.. autoclass:: kombu.Producer
|
|
:noindex:
|
|
:members:
|