This document describes the current stable version of Celery (5.2). For development docs, go here.

celery.worker.consumer

Worker consumer.

class celery.worker.consumer.Agent(c, **kwargs)

Agent starts cell actors.

conditional = True

create(c)

Create the step.

name = 'celery.worker.consumer.agent.Agent'

requires = (step:celery.worker.consumer.connection.Connection{()},)

class celery.worker.consumer.Connection(c, **kwargs)

Service managing the consumer broker connection.

info(c)

name = 'celery.worker.consumer.connection.Connection'

shutdown(c)

start(c)

class celery.worker.consumer.Consumer(on_task_request, init_callback=<function noop>, hostname=None, pool=None, app=None, timer=None, controller=None, hub=None, amqheartbeat=None, worker_options=None, disable_rate_limits=False, initial_prefetch_count=2, prefetch_multiplier=1, **kwargs)

Consumer blueprint.

class Blueprint(steps=None, name=None, on_start=None, on_close=None, on_stopped=None)

Consumer blueprint.

default_steps = ['celery.worker.consumer.connection:Connection', 'celery.worker.consumer.mingle:Mingle', 'celery.worker.consumer.events:Events', 'celery.worker.consumer.gossip:Gossip', 'celery.worker.consumer.heart:Heart', 'celery.worker.consumer.control:Control', 'celery.worker.consumer.tasks:Tasks', 'celery.worker.consumer.consumer:Evloop', 'celery.worker.consumer.agent:Agent']

name = 'Consumer'

shutdown(parent)

Strategies

alias of dict

add_task_queue(queue, exchange=None, exchange_type=None, routing_key=None, **options)

apply_eta_task(task)

Method called by the timer to apply a task with an ETA/countdown.

bucket_for_task(type)

call_soon(p, *args, **kwargs)

cancel_task_queue(queue)

connect()

Establish the broker connection used for consuming tasks.

Retries establishing the connection if the broker_connection_retry setting is enabled

connection_for_read(heartbeat=None)

connection_for_write(heartbeat=None)

create_task_handler(promise=<class 'vine.promises.promise'>)

ensure_connected(conn)

init_callback = None

Optional callback called the first time the worker is ready to receive tasks.

loop_args()

on_close()

on_connection_error_after_connected(exc)

on_connection_error_before_connected(exc)

on_decode_error(message, exc)

Callback called if an error occurs while decoding a message.

Simply logs the error and acknowledges the message so it doesn’t enter a loop.

Parameters:

• message (kombu.Message) – The message received.

• exc (Exception) – The exception being handled.

on_invalid_task(body, message, exc)

on_ready()

on_send_event_buffered()

on_unknown_message(body, message)

on_unknown_task(body, message, exc)

perform_pending_operations()

pool = None

The current worker pool instance.

register_with_event_loop(hub)

reset_rate_limits()

restart_count = -1

shutdown()

start()

stop()

timer = None

A timer used for high-priority internal tasks, such as sending heartbeats.

update_strategies()

class celery.worker.consumer.Control(c, **kwargs)

Remote control command service.

include_if(c)

Return true if bootstep should be included.

You can define this as an optional predicate that decides whether this step should be created.

name = 'celery.worker.consumer.control.Control'

requires = (step:celery.worker.consumer.tasks.Tasks{(step:celery.worker.consumer.mingle.Mingle{(step:celery.worker.consumer.events.Events{(step:celery.worker.consumer.connection.Connection{()},)},)},)},)

class celery.worker.consumer.Events(c, task_events=True, without_heartbeat=False, without_gossip=False, **kwargs)

Service used for sending monitoring events.

name = 'celery.worker.consumer.events.Events'

requires = (step:celery.worker.consumer.connection.Connection{()},)

shutdown(c)

start(c)

stop(c)

class celery.worker.consumer.Gossip(c, without_gossip=False, interval=5.0, heartbeat_interval=2.0, **kwargs)

Bootstep consuming events from other workers.

This keeps the logical clock value up to date.

call_task(task)

compatible_transport(app)

compatible_transports = {'amqp', 'redis'}

election(id, topic, action=None)

get_consumers(channel)

label = 'Gossip'

name = 'celery.worker.consumer.gossip.Gossip'

on_elect(event)

on_elect_ack(event)

on_message(prepare, message)

on_node_join(worker)

on_node_leave(worker)

on_node_lost(worker)

periodic()

register_timer()

requires = (step:celery.worker.consumer.mingle.Mingle{(step:celery.worker.consumer.events.Events{(step:celery.worker.consumer.connection.Connection{()},)},)},)

start(c)

class celery.worker.consumer.Heart(c, without_heartbeat=False, heartbeat_interval=None, **kwargs)

Bootstep sending event heartbeats.

This service sends a worker-heartbeat message every n seconds.

Note

Not to be confused with AMQP protocol level heartbeats.

name = 'celery.worker.consumer.heart.Heart'

requires = (step:celery.worker.consumer.events.Events{(step:celery.worker.consumer.connection.Connection{()},)},)

shutdown(c)

start(c)

stop(c)

class celery.worker.consumer.Mingle(c, without_mingle=False, **kwargs)

Bootstep syncing state with neighbor workers.

At startup, or upon consumer restart, this will:

• Sync logical clocks.

• Sync revoked tasks.

compatible_transport(app)

compatible_transports = {'amqp', 'redis'}

label = 'Mingle'

name = 'celery.worker.consumer.mingle.Mingle'

on_clock_event(c, clock)

on_node_reply(c, nodename, reply)

on_revoked_received(c, revoked)

requires = (step:celery.worker.consumer.events.Events{(step:celery.worker.consumer.connection.Connection{()},)},)

send_hello(c)

start(c)

sync(c)

sync_with_node(c, clock=None, revoked=None, **kwargs)

class celery.worker.consumer.Tasks(c, **kwargs)

Bootstep starting the task message consumer.

info(c)

Return task consumer info.

name = 'celery.worker.consumer.tasks.Tasks'

requires = (step:celery.worker.consumer.mingle.Mingle{(step:celery.worker.consumer.events.Events{(step:celery.worker.consumer.connection.Connection{()},)},)},)

shutdown(c)

Shutdown task consumer.

start(c)

Start task consumer.

stop(c)

Stop task consumer.