consumer.py 文件源码

def _schedule_auto_flush(self):
        """
        Schedule an automatic flush of the current batch.
        """

        if not self.auto_flush_timeout:
            return  # Auto-flush is disabled.

        self.state_lock.acquire()
        try:
            if self.flush_timer:
                return

            self.flush_timer = Timer(self.auto_flush_timeout, self.flush)
            self.flush_timer.daemon = True
            self.flush_timer.start()
        finally:
            self.state_lock.release()