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

# `celery.worker.consumer.consumer`

Worker Consumer Blueprint.

This module contains the components responsible for consuming messages from the broker, processing the messages and keeping the broker connections up and running.

class celery.worker.consumer.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.delayed_delivery:DelayedDelivery', '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.

  broker_connection_retry_attempt = 0

    Counter to track number of conn retry attempts to broker. Will be reset to 0 once successful

  bucket_for_task(type)
  call_soon(p, *args, **kwargs)
  cancel_active_requests()

    Cancel active requests during shutdown.

    Cancels all active requests that either do not require late acknowledgments or, if they do, have not been acknowledged yet.

    Does not cancel successful tasks, even if they have not been acknowledged yet.

  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(url=None, heartbeat=None)
  create_task_handler(promise=<class 'vine.promises.promise'>)
  ensure_connected(conn)
  first_connection_attempt = True

    This flag will be turned off after the first failed connection attempt.

  init_callback = None

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

  loop_args()
  property max_prefetch_count
  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.consumer.Evloop(parent, **kwargs)

  Event loop service.

  Note

  This is always started last.

  label = 'event loop'
  last = True
  name = 'celery.worker.consumer.consumer.Evloop'
  patch_all(c)
  start(c)

celery.worker.consumer.consumer.dump_body(m, body)

  Format message body for debugging purposes.