kombu
kombu is the messaging layer that Celery is built on, published separately so you can use it without Celery. It gives you an AMQP-shaped Python API (Connection, Exchange, Queue, Producer, Consumer) and then implements that API across a pile of brokers through pluggable transports. RabbitMQ and Qpid are native AMQP. Redis, Amazon SQS, MongoDB, ZooKeeper, Google Cloud Pub/Sub, Azure Service Bus, Azure Storage Queues, Kafka, and an in-process memory transport are virtual transports, meaning kombu emulates AMQP exchange and routing semantics on top of whatever primitives the broker actually has. On top of the transports it handles serialization (JSON by default, with pickle refused unless you turn it on), optional compression, connection and channel error recovery with retry policies, and connection pooling. Switching brokers is meant to be a change of URL scheme.
If you need AMQP semantics in Python and any chance of changing brokers later, kombu is the right layer and its connection recovery is better than what you would write. If you only ever talk to one broker and only need to enqueue work, you are paying an abstraction tax for portability you will never spend.
Use it if
- You want your messaging code to outlive your choice of broker. The same Producer and Consumer code runs against amqp:// in production, sqs:// in another account, and memory:// in your test suite
- You need real AMQP semantics in Python: topic exchanges with wildcard routing keys, fanout, bindings, per-message priorities, publisher confirms, and manual acknowledgement
- You want connection recovery you did not have to write. Connection.ensure() and Connection.autoretry() wrap a callable and retry it across connection and channel errors with a configurable interval and max_retries, reviving the channel between attempts
- You are already running Celery and need to publish or consume outside a task, for example feeding an external system or draining a queue from a management command. Kombu is already installed and shares the connection settings
- You want an in-process broker for tests. Connection('memory://') is a working transport, so integration tests for your producers and consumers need no Docker and no network
- You just want to put a message on Redis or RabbitMQ. Kombu makes you learn exchanges, routing keys, bindings, declaration order, and a drain_events loop before the first message moves. pika against RabbitMQ, or the redis client against a list, is far fewer concepts for a simple producer
- You want a task queue. Kombu moves bytes; it has no workers, no retries with backoff on your business logic, no scheduling, no result backend, and no monitoring. That is Celery, or dramatiq if you want something smaller
- You are on a virtual transport and assume AMQP features come with it. The README's own transport table is blunt about the gaps: SQS has no priority and no message TTL, ZooKeeper and the in-memory and SLMQ and Pyro transports have no fanout, and for several virtual transports the exchange and queue declarations exist only in the memory of the client process, so every producer and consumer has to declare them independently
- Your application is asyncio. Kombu is a blocking library with its own event loop hub built for Celery's worker; using it from async code means a thread pool. aio-pika is the native answer for RabbitMQ under asyncio
- You are sensitive to dependency pinning. kombu pins vine to exactly 5.1.0 and the redis extra carries an exclusion list of specific redis versions. In a large application this is the kind of constraint that blocks an unrelated upgrade
- You want documentation that teaches you the operational parts. The Sphinx reference covers the API well, but the things that actually break in production (Redis visibility timeout and unacked message redelivery, prefetch tuning, what happens to declarations after a reconnect) mostly live in the Celery docs and in issue threads. There are 161 open issues (235 counting PRs)
Setup reality
pip install kombu gives you four dependencies (amqp, vine pinned to exactly 5.1.0, tzdata, packaging) and exactly one usable broker: RabbitMQ or another AMQP server, over py-amqp. Everything else is an extra, and the extras are heavier than people expect. pip install 'kombu[redis]' adds redis with a version exclusion list; 'kombu[sqs]' adds boto3, urllib3 and, on non-Windows CPython, pycurl, which needs libcurl development headers and a compiler when no matching wheel exists; 'kombu[gcpubsub]' drags in grpcio and protobuf at pinned versions; 'kombu[confluentkafka]' brings librdkafka. Python 3.9 or newer is required. Choosing the transport is done through the URL scheme (amqp://, redis://, sqs://, mongodb://, memory://), not through an import, so a typo in the scheme surfaces as a transport lookup error rather than a connection error. Two habits will save you time. Declare your queues from both the producer and the consumer by passing declare=[queue] to publish, because on virtual transports the topology only exists in whichever process created it. And know that serialization defaults are strict: JSON is the default and decodes without configuration, while a message published with serializer='pickle' raises kombu.exceptions.ContentDisallowed on the consumer side until you explicitly enable insecure serializers or pass accept=['pickle']. That is a security default, not a bug.
Patterns
Connect and publish to an exchangepublish-a-message
from kombu import Connection, Exchange, Queue
media = Exchange('media', 'direct', durable=True)
video_queue = Queue('video', exchange=media, routing_key='video')
with Connection('amqp://guest:guest@localhost//') as conn:
producer = conn.Producer(serializer='json')
producer.publish(
{'name': '/tmp/clip.avi', 'size': 1301013},
exchange=media,
routing_key='video',
declare=[video_queue],
delivery_mode=2, # persist to disk
)declare=[video_queue] makes the producer create the queue and its binding if they are missing, which matters because a message published to an exchange with no bound queue is dropped without an error. delivery_mode=2 is the persistent setting; the default is in-memory only, so a broker restart loses the message.
Consume with a drain_events loopconsume-messages
from kombu import Connection, Exchange, Queue
media = Exchange('media', 'direct', durable=True)
video_queue = Queue('video', exchange=media, routing_key='video')
def handle(body, message):
try:
process(body)
message.ack()
except Exception:
message.reject(requeue=False) # send to DLX or drop
with Connection('amqp://guest:guest@localhost//') as conn:
with conn.Consumer(video_queue, accept=['json'], callbacks=[handle],
prefetch_count=10):
while True:
conn.drain_events(timeout=None)The callback signature is always (body, message) and you must call ack, reject, or requeue on every path; an unacked message is redelivered when the connection drops, so a handler that returns without acking produces an infinite redelivery loop. prefetch_count caps how many unacked messages the broker will hand you at once, and leaving it unset means one slow consumer can hoard the whole queue.
Run a long-lived worker with ConsumerMixinconsumer-worker
from kombu.mixins import ConsumerMixin
class Worker(ConsumerMixin):
def __init__(self, connection, queues):
self.connection = connection
self.queues = queues
def get_consumers(self, Consumer, channel):
return [Consumer(queues=self.queues, accept=['json'],
callbacks=[self.on_message], prefetch_count=10)]
def on_message(self, body, message):
process(body)
message.ack()
with Connection('amqp://guest:guest@localhost//', heartbeat=10) as conn:
Worker(conn, [video_queue]).run()ConsumerMixin owns the loop, reconnects with backoff, and redeclares your consumers after a reconnect, which is the part a hand-written drain_events loop always gets wrong. Set heartbeat on the Connection or a firewall will silently drop an idle connection and your worker will sit there consuming nothing.
Use the plain queue API when you do not need AMQPsimple-queue
from kombu import Connection
with Connection('amqp://guest:guest@localhost//') as conn:
queue = conn.SimpleQueue('jobs')
queue.put({'id': 7}, serializer='json')
print(queue.qsize())
message = queue.get(block=True, timeout=5)
print(message.payload)
message.ack()
queue.close()
# SimpleBuffer is the same API with durability turned off
buffer = conn.SimpleBuffer('scratch')SimpleQueue creates a durable direct exchange and queue with the same name and hides the routing entirely, which is the shortest path when portability across brokers is the only reason you chose kombu. get() raises queue.Empty on timeout, and you still have to ack. Close it yourself or the channel leaks.
Route with a topic exchange and wildcardstopic-routing
from kombu import Exchange, Queue
events = Exchange('events', 'topic', durable=True)
all_orders = Queue('orders.all', exchange=events, routing_key='order.#')
failures = Queue('orders.failed', exchange=events, routing_key='order.*.failed')
producer.publish(payload, exchange=events,
routing_key='order.eu.failed',
declare=[all_orders, failures])In topic routing, * matches exactly one dot-separated word and # matches zero or more, so order.# catches order.eu.failed but order.*.failed does not catch order.failed. Both queues here receive the message; a topic exchange fans out to every binding that matches, it does not stop at the first.
Retry across connection and channel errorsconnection-recovery
from kombu import Connection
conn = Connection('amqp://guest:guest@localhost//')
conn.ensure_connection(max_retries=3, interval_start=0,
interval_step=2, interval_max=10)
def send(payload, channel):
producer = conn.Producer(channel)
producer.publish(payload, exchange=media, routing_key='video',
declare=[video_queue])
safe_send = conn.autoretry(send, max_retries=3)
safe_send({'id': 1})
# publish-level retry policy, no wrapper needed
producer.publish(payload, retry=True,
retry_policy={'max_retries': 3, 'interval_start': 0,
'interval_step': 1, 'interval_max': 5})autoretry passes a fresh channel into your function on every attempt, which is why the callable has to take a channel argument; a closure over a stale channel is the usual reason retries keep failing. Retrying a publish is at-least-once, so the consumer must be idempotent.
Control serialization and what you acceptserialization-security
producer.publish(payload, serializer='json', compression='gzip')
# consumers declare what they will deserialize
conn.Consumer(video_queue, accept=['json'], callbacks=[handle])
# pickle raises kombu.exceptions.ContentDisallowed by default
from kombu.serialization import enable_insecure_serializers
enable_insecure_serializers(['pickle']) # only if you mustVerified against 5.6.2: publishing with serializer='pickle' and consuming it raises ContentDisallowed ('Refusing to deserialize disabled content of type pickle'), because a pickle payload from an untrusted queue is remote code execution. JSON decodes without configuring accept, but naming it explicitly documents the contract and stops a future default from widening it.
Point the same code at Redisredis-transport
from kombu import Connection
conn = Connection(
'redis://localhost:6379/0',
transport_options={
'visibility_timeout': 3600, # seconds before redelivery
'fanout_prefix': True,
'fanout_patterns': True,
},
)Redis has no acknowledgement concept, so kombu emulates it: a delivered message is hidden for visibility_timeout seconds and redelivered if not acked in time. Any task that legitimately runs longer than that gets processed twice, which is the single most common Redis-transport incident. Set it above your slowest handler.
Run against Amazon SQSsqs-transport
from kombu import Connection
conn = Connection(
'sqs://', # credentials from the environment or IAM role
transport_options={
'region': 'ap-south-1',
'visibility_timeout': 3600,
'polling_interval': 1,
'predefined_queues': {
'jobs': {'url': 'https://sqs.ap-south-1.amazonaws.com/123/jobs'},
},
},
)SQS supports neither message priority nor TTL, and fanout only works through SNS after you enable the supports_fanout option, which changes your AWS bill. polling_interval controls how often an empty queue is polled, and every poll is a billable API call, so a short interval across many workers costs real money.
Give some messages priority on RabbitMQpriority-queues
from kombu import Exchange, Queue
jobs = Queue(
'jobs',
exchange=Exchange('jobs', 'direct'),
routing_key='jobs',
queue_arguments={'x-max-priority': 10, 'x-message-ttl': 86400000},
)
producer.publish(payload, exchange=jobs.exchange,
routing_key='jobs', priority=5, declare=[jobs])x-max-priority is set at declaration time and cannot be changed later; adding it to an existing queue means deleting and recreating it, which loses the messages. Priority only orders what is already sitting in the queue, so with a high prefetch_count your consumers have already grabbed the low-priority backlog and the setting appears to do nothing.
Share connections and producers across threadsconnection-pools
from kombu import Connection
from kombu.pools import connections, producers, set_limit
set_limit(20)
conn = Connection('amqp://guest:guest@localhost//')
with producers[conn].acquire(block=True, timeout=5) as producer:
producer.publish(payload, exchange=media, routing_key='video',
declare=[video_queue])
with connections[conn].acquire(block=True) as acquired:
acquired.default_channel.queue_purge('video')The pools are keyed on the connection object, so building a new Connection per request defeats them entirely; create it once at module level. The global limit defaults to 200 and is per process, and after os.fork the pools are reset, which is why a pre-fork web server needs the connection created in the worker rather than the parent.
Test producers and consumers without a brokertest-with-memory-transport
from kombu import Connection, Exchange, Queue
def test_video_pipeline():
ex = Exchange('media', 'direct')
q = Queue('video', exchange=ex, routing_key='video')
received = []
with Connection('memory://') as conn:
conn.Producer(serializer='json').publish(
{'name': 'x.avi'}, exchange=ex, routing_key='video', declare=[q])
with conn.Consumer(q, accept=['json'],
callbacks=[lambda b, m: (received.append(b), m.ack())]):
conn.drain_events(timeout=2)
assert received == [{'name': 'x.avi'}]Verified working on 5.6.2. The memory transport is in-process and per-Connection, so a producer and consumer in different Connection objects will not see each other. It also has no fanout support, so tests that depend on fanout routing pass locally against RabbitMQ and fail here for the wrong reason.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| celery | PyPI | You actually want a task queue with workers, retries, scheduling, and results rather than a raw messaging layer |
| pika | PyPI | You only target RabbitMQ and want a thin AMQP 0.9.1 client with no abstraction layer between you and the protocol |
| aio-pika | PyPI | Your application is asyncio end to end and you need a native async RabbitMQ client instead of running blocking calls in a thread |
| dramatiq | PyPI | You want background jobs on RabbitMQ or Redis with far less surface area and configuration than Celery |