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

# Source code for celery.worker.heartbeat

    """Heartbeat service.

    This is the internal thread responsible for sending heartbeat events
    at regular intervals (may not be an actual thread).
    """
    from celery.signals import heartbeat_sent
    from celery.utils.sysinfo import load_average

    from .state import SOFTWARE_INFO, active_requests, all_total_count

    __all__ = ('Heart',)

    class Heart:
        """Timer sending heartbeats at regular intervals.

        Arguments:
            timer (kombu.asynchronous.timer.Timer): Timer to use.
            eventer (celery.events.EventDispatcher): Event dispatcher
                to use.
            interval (float): Time in seconds between sending
                heartbeats.  Default is 2 seconds.
        """

        def __init__(self, timer, eventer, interval=None):
            self.timer = timer
            self.eventer = eventer
            self.interval = float(interval or 2.0)
            self.tref = None

            # Make event dispatcher start/stop us when enabled/disabled.
            self.eventer.on_enabled.add(self.start)
            self.eventer.on_disabled.add(self.stop)

            # Only send heartbeat_sent signal if it has receivers.
            self._send_sent_signal = (
                heartbeat_sent.send if heartbeat_sent.receivers else None)

        def _send(self, event, retry=True):
            if self._send_sent_signal is not None:
                self._send_sent_signal(sender=self)
            return self.eventer.send(event, freq=self.interval,
                                     active=len(active_requests),
                                     processed=all_total_count[0],
                                     loadavg=load_average(),
                                     retry=retry,
                                     **SOFTWARE_INFO)

        def start(self):
            if self.eventer.enabled:
                self._send('worker-online')
                self.tref = self.timer.call_repeatedly(
                    self.interval, self._send, ('worker-heartbeat',),
                )

        def stop(self):
            if self.tref is not None:
                self.timer.cancel(self.tref)
                self.tref = None
            if self.eventer.enabled:
                self._send('worker-offline', retry=False)