This document is for Kombu's development version, which can be significantly different from previous releases. Get the stable docs here: 5.3.

Redis Transport - kombu.transport.redis

Redis transport module for Kombu.

Features

• Type: Virtual

• Supports Direct: Yes

• Supports Topic: Yes

• Supports Fanout: Yes

• Supports Priority: Yes

• Supports Queue TTL: Yes

• Supports Message TTL: No

Connection String

Connection string has the following format:

redis://[USER:PASSWORD@]REDIS_ADDRESS[:PORT][/VIRTUALHOST]
rediss://[USER:PASSWORD@]REDIS_ADDRESS[:PORT][/VIRTUALHOST]

To use sentinel for dynamic Redis discovery, the connection string has following format:

sentinel://[USER:PASSWORD@]SENTINEL_ADDRESS[:PORT]

Transport Options

• sep

• ack_emulation: (bool) If set to True transport will simulate Acknowledge of AMQP protocol.

• unacked_key

• unacked_index_key

• unacked_mutex_key

• unacked_mutex_expire

• visibility_timeout

• unacked_restore_limit

• unacked_restore_interval: (int) Seconds between periodic restore_visible sweeps in the async (event loop / prefork) path. Defaults to 10. Lower this to recover abandoned messages faster when using a low visibility_timeout.

• unacked_restore_throttle: (int) Only run an actual Redis scan on every Nth restore_visible call. Defaults to 10. The effective async sweep period is roughly unacked_restore_interval * unacked_restore_throttle seconds, so set this to 1 to make unacked_restore_interval the sole control.

• fanout_prefix

• fanout_patterns

• global_keyprefix: (str) The global key prefix to be prepended to all keys used by Kombu

• socket_timeout

• socket_connect_timeout

• socket_keepalive

• socket_keepalive_options

• queue_order_strategy

• max_connections

• health_check_interval

• reauth_check_interval: (int) How often, in seconds, to flush pending streaming re-authentication tokens (emitted by a redis.credentials.StreamingCredentialProvider such as the Entra ID / IAM providers) onto the long-lived BRPOP and pub/sub connections held by the transport. Defaults to 10. See Channel.maybe_reauth().

• retry_on_timeout

• priority_steps

• client_name: (str) The name to use when connecting to Redis server.

Queue Arguments

• x-expires: (int) Time in milliseconds for queues to expire if there’s no activity. The queue will be automatically deleted after this period of inactivity. This is a per-queue argument and should be supplied via Queue(expires=...) or Queue(..., queue_arguments={'x-expires': ...}) rather than as a connection-level transport option.

Transport

class kombu.transport.redis.Transport(*args, **kwargs)

Redis Transport.

class Channel(*args, **kwargs)

Redis Channel.

class QoS(*args, **kwargs)

Redis Ack Emulation.

ack(delivery_tag)

Acknowledge message and remove from transactional state.

append(message, delivery_tag)

Append message to transactional state.

pipe_or_acquire(pipe=None, client=None)

reject(delivery_tag, requeue=False)

Remove from transactional state and requeue message.

restore_at_shutdown = True

If disabled, unacked messages won’t be restored at shutdown.

restore_by_tag(tag, client=None, leftmost=False)

restore_unacked(client=None)

Restore all unacknowledged messages.

restore_visible(start=0, num=10, interval=10)

Restore any pending unacknowledged messages.

To be filled in for visibility_timeout style implementations.

Note:

This is implementation optional, and currently only used by the Redis transport.

property unacked_index_key

property unacked_key

property unacked_mutex_expire

property unacked_mutex_key

property visibility_timeout

ack_emulation = True

property active_queues

List of queues being consumed from (excluding fanout queues).

property async_pool

basic_cancel(consumer_tag)

Cancel consumer by consumer tag.

basic_consume(queue, *args, **kwargs)

Consume from queue.

property client

Client used to publish messages, BRPOP etc.

client_name = None

close()

Close channel.

Cancel all consumers, and requeue unacked messages.

conn_or_acquire(client=None)

connection_class

alias of Connection

connection_class_ssl

alias of SSLConnection

create_publish_batch(max_size)

Create a producer-scoped Redis publish batch.

fanout_patterns = True

If enabled the fanout exchange will support patterns in routing and binding keys (like a topic exchange but using PUB/SUB).

Enabled by default since Kombu 4.x. Disable for backwards compatibility with Kombu 3.x.

fanout_prefix = True

Transport option to disable fanout keyprefix. Can also be string, in which case it changes the default prefix (‘/{db}.’) into to something else. The prefix must include a leading slash and a trailing dot.

Enabled by default since Kombu 4.x. Disable for backwards compatibility with Kombu 3.x.

from_transport_options = ('body_encoding', 'deadletter_queue', 'sep', 'ack_emulation', 'unacked_key', 'unacked_index_key', 'unacked_mutex_key', 'unacked_mutex_expire', 'visibility_timeout', 'unacked_restore_limit', 'unacked_restore_interval', 'unacked_restore_throttle', 'fanout_prefix', 'fanout_patterns', 'global_keyprefix', 'socket_timeout', 'socket_connect_timeout', 'socket_keepalive', 'socket_keepalive_options', 'queue_order_strategy', 'max_connections', 'health_check_interval', 'retry_on_timeout', 'priority_steps', 'client_name')

get_table(exchange)

Get table of bindings for exchange.

global_keyprefix = ''

The global key prefix will be prepended to all keys used by Kombu, which can be useful when a redis database is shared by different users. By default, no prefix is prepended.

health_check_interval = 25

keyprefix_fanout = '/{db}.'

keyprefix_queue = '_kombu.binding.%s'

max_connections = 10

maybe_reauth()

Flush pending streaming re-auth tokens onto long-lived connections.

The transport holds two connections for the lifetime of the worker that are never released back to the pool: the BRPOP connection (used to consume from ordinary queues) and the pub/sub LISTEN connection (used to consume from fanout queues). Because they are never released, redis-py’s release-triggered re-authentication never fires for them, so a streaming credential provider’s rotated tokens never reach them and the broker eventually severs the connections when the original credentials expire (e.g. the 12h limit imposed by AWS ElastiCache with IAM auth).

This is called periodically from the event loop (a single thread), so it can safely flush the tokens without racing the socket reads.

property pool

prepare_queue_arguments(arguments, **kwargs)

priority(n)

priority_steps = [0, 3, 6, 9]

queue_order_strategy = 'round_robin'

Order in which we consume from queues.

Can be either string alias, or a cycle strategy class

• round_robin (round_robin_cycle).

  Make sure each queue has an equal opportunity to be consumed from.

• sorted (sorted_cycle).

  Consume from queues in alphabetical order. If the first queue in the sorted list always contains messages, then the rest of the queues will never be consumed from.

• priority (priority_cycle).

  Consume from queues in original order, so that if the first queue always contains messages, the rest of the queues in the list will never be consumed from.

The default is to consume from queues in round robin.

retry_on_timeout = None

sep = '\x06\x16'

socket_connect_timeout = None

socket_keepalive = None

socket_keepalive_options = None

socket_timeout = None

property subclient

Pub/Sub connection used to consume fanout queues.

property supports_batch_publish

Return whether pipelines can run without ambiguous write replay.

supports_fanout = True

flag set if the channel supports fanout exchanges.

unacked_index_key = 'unacked_index'

unacked_key = 'unacked'

unacked_mutex_expire = 300

unacked_mutex_key = 'unacked_mutex'

unacked_restore_interval = 10

Seconds between periodic restore_visible sweeps in the async (event loop / prefork) path. Lower this to recover abandoned messages faster when using a low visibility_timeout.

unacked_restore_limit = None

unacked_restore_throttle = 10

Only run an actual Redis scan on every Nth restore_visible call. Protects the synchronous poll path (restore is attempted on every empty poll) from hitting Redis too often. Set to 1 to scan on every call. Effective async sweep period is roughly unacked_restore_interval * unacked_restore_throttle seconds.

visibility_timeout = 3600

brpop_timeout = 1

channel_errors = (<class 'amqp.exceptions.ChannelError'>, <class 'redis.exceptions.DataError'>, <class 'redis.exceptions.InvalidResponse'>, <class 'redis.exceptions.ResponseError'>)

Tuple of errors that can happen due to channel/method failure.

connection_errors = (<class 'amqp.exceptions.ConnectionError'>, <class 'kombu.exceptions.InconsistencyError'>, <class 'OSError'>, <class 'OSError'>, <class 'OSError'>, <class 'redis.exceptions.ConnectionError'>, <class 'redis.exceptions.BusyLoadingError'>, <class 'redis.exceptions.AuthenticationError'>, <class 'redis.exceptions.TimeoutError'>, <class 'KeyError'>)

Tuple of errors that can happen due to connection failure.

default_port = 6379

port number used when no port is specified.

driver_name = 'redis'

Name of driver library (e.g. ‘py-amqp’, ‘redis’).

driver_type = 'redis'

Type of driver, can be used to separate transports using the AMQP protocol (driver_type: ‘amqp’), Redis (driver_type: ‘redis’), etc…

driver_version()

implements = {'asynchronous': True, 'batch_publish': True, 'exchange_type': frozenset({'direct', 'fanout', 'topic'}), 'heartbeats': False}

on_readable(fileno)

Handle AIO event for one of our file descriptors.

polling_interval = None

Time to sleep between unsuccessful polls.

register_with_event_loop(connection, loop)

property supports_batch_publish

Return whether configured timeout handling permits safe batching.

Transport-specific notes

Added in version 5.6.0.

Redis now honours the generic polling_interval option (present in SQS, etcd, Zookeeper, …). When you pass

app.conf.broker_transport_options = {"polling_interval": 10}

the worker uses that value as the timeout for the underlying BRPOP call, so it issues at most one poll every 10 seconds while the queue is empty. The default remains 1 second to stay backward-compatible.

Added in version 5.7.0.

Supports Queue TTL

Streaming credentials / automatic re-authentication

Added in version 5.7.0.

The transport supports rotating credentials supplied by a redis.credentials.StreamingCredentialProvider (for example the Microsoft Entra ID provider from redis-entraid, or an AWS ElastiCache IAM provider). Configure it like any other credential provider:

app.conf.broker_transport_options = {
    "visibility_timeout": 30,
    "unacked_restore_interval": 5,
    "unacked_restore_throttle": 1,
}

Note

unacked_restore_interval only affects the asynchronous path (a prefork worker driven by the event loop on Linux/macOS). On the synchronous path — used by eventlet/gevent pools, on Windows, or by any plain connection.drain_events() loop — there is no restore timer; a restore is attempted on every empty poll (~``brpop_timeout``, 1 second) and only unacked_restore_throttle governs how often Redis is actually scanned.

Warning

restore_visible uses the scanning worker’s own visibility_timeout as the cutoff — the sorted set stores only the fetch timestamp, not each message’s timeout. Configure the same visibility_timeout on every worker that may run the sweep, otherwise restore timing becomes inconsistent: too low re-queues messages that are still being processed (duplicate execution), too high delays recovery.

Dedicated “janitor” worker for green pools

Under the eventlet and gevent pools every greenlet runs cooperatively in a single OS thread. A CPU-bound task that does not yield blocks the hub, which also stalls the broker consumer loop — so restore sweeps (and heartbeats) stop running until the task finishes. Visibility timeouts may then expire without anything re-queuing the affected messages.

Because the unacked index is global, you can side-step this by running a small dedicated prefork worker whose only job is to drive restores. Point it at the same Redis database and global_keyprefix as your real workers, have it consume a throwaway queue (so active_queues is non-empty and the restore loop fires), give it the same visibility_timeout as the rest of the fleet, and tune its restore cadence aggressively:

# janitor_app.py
from celery import Celery

app = Celery("janitor", broker="redis://localhost:6379/0")
app.conf.broker_transport_options = {
    "visibility_timeout": 3600,        # MUST match the real workers
    "unacked_restore_interval": 5,
    "unacked_restore_throttle": 1,
}

$ celery -A janitor_app worker -P prefork -c 1 -Q janitor_dummy

The janitor never needs to consume the real queues: restore_visible re-routes each recovered message to its original queue using the bindings stored in Redis. Its hub is independent of the green-pool workers, so it keeps reaping expired messages even while they are busy with CPU-bound work — and it doubles as a safety net for messages stranded by crashed workers.

Note

The janitor only drives the sweep; the per-message restore work is unchanged, and all workers still contend for the same unacked_mutex so only one sweep runs at a time. ack_emulation must stay enabled for the unacked index to exist.

“credential_provider”: my_streaming_credential_provider,

}

Such providers emit a fresh authentication token in the background before the current one expires. redis-py delivers these tokens to pooled connections when they are released back to the pool, but the Redis transport keeps two connections busy for the entire lifetime of the worker — the BRPOP connection (used to consume regular queues) and the pub/sub LISTEN connection (used to consume fanout queues) — so they are never released and would otherwise never receive a rotated token. The broker then severs them once the original credentials expire (e.g. AWS ElastiCache with IAM auth enforces a hard 12-hour connection limit), causing redelivered messages, interrupted in-flight tasks and brief worker unavailability.

To avoid this, the transport periodically flushes any pending token onto those long-lived connections from the event loop:

• the BRPOP connection is re-authenticated in place with an AUTH command, sent only when no blocking pop is in flight;

• the pub/sub LISTEN connection cannot process AUTH while subscribed under RESP2, so it is transparently reconnected (and re-subscribed) to pick up the new credentials. Under RESP3, redis-py re-authenticates pub/sub connections itself and the transport leaves them untouched.

How often the flush runs is controlled by the reauth_check_interval transport option (seconds, default 10):

app.conf.broker_transport_options = {
    "credential_provider": my_streaming_credential_provider,
    "reauth_check_interval": 10,
}

When no streaming credential provider is configured this machinery is a cheap no-op, so it is always safe to leave enabled.

Queue arguments

The following queue argument is supported. Pass it per-queue via Queue(expires=...) or Queue(..., queue_arguments={'x-expires': ...}), not as a connection-level transport option.

x-expires (int)

Time in milliseconds for the queue to expire if there is no activity. The queue will be automatically deleted after this period of inactivity.

Channel

class kombu.transport.redis.Channel(*args, **kwargs)

Redis Channel.

class QoS(*args, **kwargs)

Redis Ack Emulation.

ack(delivery_tag)

Acknowledge message and remove from transactional state.

append(message, delivery_tag)

Append message to transactional state.

pipe_or_acquire(pipe=None, client=None)

reject(delivery_tag, requeue=False)

Remove from transactional state and requeue message.

restore_at_shutdown = True

If disabled, unacked messages won’t be restored at shutdown.

restore_by_tag(tag, client=None, leftmost=False)

restore_unacked(client=None)

Restore all unacknowledged messages.

restore_visible(start=0, num=10, interval=10)

Restore any pending unacknowledged messages.

To be filled in for visibility_timeout style implementations.

Note:

This is implementation optional, and currently only used by the Redis transport.

property unacked_index_key

property unacked_key

property unacked_mutex_expire

property unacked_mutex_key

property visibility_timeout

ack_emulation = True

property active_queues

List of queues being consumed from (excluding fanout queues).

property async_pool

basic_cancel(consumer_tag)

Cancel consumer by consumer tag.

basic_consume(queue, *args, **kwargs)

Consume from queue.

property client

Client used to publish messages, BRPOP etc.

client_name = None

close()

Close channel.

Cancel all consumers, and requeue unacked messages.

conn_or_acquire(client=None)

connection_class

alias of Connection

connection_class_ssl

alias of SSLConnection

create_publish_batch(max_size)

Create a producer-scoped Redis publish batch.

fanout_patterns = True

If enabled the fanout exchange will support patterns in routing and binding keys (like a topic exchange but using PUB/SUB).

Enabled by default since Kombu 4.x. Disable for backwards compatibility with Kombu 3.x.

fanout_prefix = True

Transport option to disable fanout keyprefix. Can also be string, in which case it changes the default prefix (‘/{db}.’) into to something else. The prefix must include a leading slash and a trailing dot.

Enabled by default since Kombu 4.x. Disable for backwards compatibility with Kombu 3.x.

from_transport_options = ('body_encoding', 'deadletter_queue', 'sep', 'ack_emulation', 'unacked_key', 'unacked_index_key', 'unacked_mutex_key', 'unacked_mutex_expire', 'visibility_timeout', 'unacked_restore_limit', 'unacked_restore_interval', 'unacked_restore_throttle', 'fanout_prefix', 'fanout_patterns', 'global_keyprefix', 'socket_timeout', 'socket_connect_timeout', 'socket_keepalive', 'socket_keepalive_options', 'queue_order_strategy', 'max_connections', 'health_check_interval', 'retry_on_timeout', 'priority_steps', 'client_name')

get_table(exchange)

Get table of bindings for exchange.

global_keyprefix = ''

The global key prefix will be prepended to all keys used by Kombu, which can be useful when a redis database is shared by different users. By default, no prefix is prepended.

health_check_interval = 25

keyprefix_fanout = '/{db}.'

keyprefix_queue = '_kombu.binding.%s'

max_connections = 10

maybe_reauth()

Flush pending streaming re-auth tokens onto long-lived connections.

The transport holds two connections for the lifetime of the worker that are never released back to the pool: the BRPOP connection (used to consume from ordinary queues) and the pub/sub LISTEN connection (used to consume from fanout queues). Because they are never released, redis-py’s release-triggered re-authentication never fires for them, so a streaming credential provider’s rotated tokens never reach them and the broker eventually severs the connections when the original credentials expire (e.g. the 12h limit imposed by AWS ElastiCache with IAM auth).

This is called periodically from the event loop (a single thread), so it can safely flush the tokens without racing the socket reads.

property pool

prepare_queue_arguments(arguments, **kwargs)

priority(n)

priority_steps = [0, 3, 6, 9]

queue_order_strategy = 'round_robin'

Order in which we consume from queues.

Can be either string alias, or a cycle strategy class

• round_robin (round_robin_cycle).

  Make sure each queue has an equal opportunity to be consumed from.

• sorted (sorted_cycle).

  Consume from queues in alphabetical order. If the first queue in the sorted list always contains messages, then the rest of the queues will never be consumed from.

• priority (priority_cycle).

  Consume from queues in original order, so that if the first queue always contains messages, the rest of the queues in the list will never be consumed from.

The default is to consume from queues in round robin.

retry_on_timeout = None

sep = '\x06\x16'

socket_connect_timeout = None

socket_keepalive = None

socket_keepalive_options = None

socket_timeout = None

property subclient

Pub/Sub connection used to consume fanout queues.

property supports_batch_publish

Return whether pipelines can run without ambiguous write replay.

supports_fanout = True

flag set if the channel supports fanout exchanges.

unacked_index_key = 'unacked_index'

unacked_key = 'unacked'

unacked_mutex_expire = 300

unacked_mutex_key = 'unacked_mutex'

unacked_restore_interval = 10

Seconds between periodic restore_visible sweeps in the async (event loop / prefork) path. Lower this to recover abandoned messages faster when using a low visibility_timeout.

unacked_restore_limit = None

unacked_restore_throttle = 10

Only run an actual Redis scan on every Nth restore_visible call. Protects the synchronous poll path (restore is attempted on every empty poll) from hitting Redis too often. Set to 1 to scan on every call. Effective async sweep period is roughly unacked_restore_interval * unacked_restore_throttle seconds.

visibility_timeout = 3600

SentinelChannel

class kombu.transport.redis.SentinelChannel(*args, **kwargs)

Channel with explicit Redis Sentinel knowledge.

Broker url is supposed to look like:

sentinel://0.0.0.0:26379;sentinel://0.0.0.0:26380/...

where each sentinel is separated by a ;.

Other arguments for the sentinel should come from the transport options (see transport_options of Connection).

You must provide at least one option in Transport options:

• master_name - name of the redis group to poll

Example:

>>> import kombu
>>> c = kombu.Connection(
     'sentinel://sentinel1:26379;sentinel://sentinel2:26379',
     transport_options={'master_name': 'mymaster'}
)
>>> c.connect()

connection_class

alias of SentinelManagedConnection

connection_class_ssl

alias of SentinelManagedSSLConnection

from_transport_options = ('body_encoding', 'deadletter_queue', 'sep', 'ack_emulation', 'unacked_key', 'unacked_index_key', 'unacked_mutex_key', 'unacked_mutex_expire', 'visibility_timeout', 'unacked_restore_limit', 'unacked_restore_interval', 'unacked_restore_throttle', 'fanout_prefix', 'fanout_patterns', 'global_keyprefix', 'socket_timeout', 'socket_connect_timeout', 'socket_keepalive', 'socket_keepalive_options', 'queue_order_strategy', 'max_connections', 'health_check_interval', 'retry_on_timeout', 'priority_steps', 'client_name', 'master_name', 'min_other_sentinels', 'sentinel_kwargs')