Consumer class provides all
the structure required for implementing a rejected consumer,
functionality designed to make writing consumers even easier. When messages
are received by consumers extending
if the message’s
content_type property contains one of the supported mime-types,
the message body will automatically be deserialized, making the deserialized
message body available via the
body attribute. Additionally, should one of
content_encoding types (
bzip2) be specified in the
message’s property, it will automatically be decoded.
Message Type Validation¶
In any of the consumer base classes, if the
MESSAGE_TYPE attribute is set,
type property of incoming messages will be validated against when a message is
received, checking for string equality against the
If they are not matched, the consumer will not process the message and will drop the
message without an exception if the
DROP_INVALID_MESSAGES attribute is set to
True. If it is
ConsumerException is raised.
Republishing of Dropped Messages¶
If the consumer is configured by specifying
DROP_EXCHANGE as an attribute of
the consumer class or in the consumer configuration with the
configuration variable, when a message is dropped, it is published to that
exchange prior to the message being rejected in RabbitMQ. When the
message is republished, four new values are added to the AMQP
X-Dropped-By header value contains the configured name of the
consumer that dropped the message.
X-Dropped-Reason contains the
reason the message was dropped (eg invalid message type or maximum error
X-Dropped-Timestamp value contains the ISO-8601 formatted
timestamp of when the message was dropped. Finally, the
X-Original-Exchange value contains the original exchange that the
message was published to.
There are three exception types that consumer applications should raise to handle problems that may arise when processing a message. When these exceptions are raised, rejected will reject the message delivery, letting RabbitMQ know that there was a failure.
ConsumerException should be
raised when there is a problem in the consumer itself, such as inability to contact
a database server or other resources. When a
ConsumerException is raised,
the message will be rejected and requeued, adding it back to the RabbitMQ it
was delivered back to. Additionally, rejected keeps track of consumer exceptions
and will shutdown the consumer process and start a new one once a consumer has
exceeded its configured maximum error count within a
60 second window. The
default maximum error count is
MessageException should be
raised when there is a problem with the message. When this exception is raised,
the message will be rejected on the RabbitMQ server without requeue, discarding
the message. This should be done when there is a problem with the message itself,
such as a malformed payload or non-supported properties like
If a consumer raises a
message that was being processed will be republished to the exchange
specified by the
error exchange configuration value or the
ERROR_EXCHANGE attribute of the consumer’s class. The message will be
published using the routing key that was last used for the message. The
original message body and properties will be used and two additional
header property values may be added:
X-Processing-Exceptioncontains the string value of the exception that was
raised, if specified.
X-Processing-Exceptions contains the quantity of processing exceptions
that have been raised for the message.
In combination with a queue that has
x-dead-letter-exchange that points to the original exchange for the
queue the consumer is consuming off of, you can implement a delayed retry
cycle for messages that are failing to process due to external resource or
ERROR_MAX_RETRY is set on the class, the headers for each method
will be inspected and if the value of
greater than or equal to the
ERROR_MAX_RETRY value, the message will
If unhandled exceptions are raised by a consumer, they will be caught by rejected, logged, and turned into a
RejectedException(value=None, metric=None, *args, **kwargs)¶
Base exception for
If provided, the metric will be used to automatically record exception metric counts using the path [prefix].[consumer-name].exceptions.[exception-type].[metric].
Positional and keyword arguments are used to format the value that is passed in when providing the string value of the exception.
New in version 3.19.0.
ConsumerException(value=None, metric=None, *args, **kwargs)¶
May be called when processing a message to indicate a problem that the Consumer may be experiencing that should cause it to stop.
MessageException(value=None, metric=None, *args, **kwargs)¶
Invoke when a message should be rejected and not re-queued, but not due to a processing error that should cause the consumer to stop.
ProcessingException(value=None, metric=None, *args, **kwargs)¶
Invoke when a message should be rejected and not re-queued, but not due to a processing error that should cause the consumer to stop. This should be used for when you want to reject a message which will be republished to a retry queue, without anything being stated about the exception.