python - threads not correctly running in Celery - Stack Overflow

admin管理员组

文章数量:1125964

here is the testing code:

@app.task(base=AbortableServerSideContextTask)
def test_watch_dog():
    from common.utils.test_with_heartbeat import TestWithWatchDog
    with TestWithWatchDog():
        times = 0
        while times < 10:
            from flask import current_app
            current_app.logger.info(f'main thread counting times:{times}')
            times += 1
            time.sleep(2) # processing

import threading

from flask import current_app


class TestWithWatchDog:
    def __init__(self):
        self.watch_dog = threading.Thread(target=self._watch_dog, daemon=False)
        self.watch_dog_shutdown_event = threading.Event()
        self.logger = current_app.logger

    def __enter__(self):
        self.logger.info("TestWithHeartBeat entered")
        self.watch_dog.start()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.watch_dog_shutdown_event.set()
        self.watch_dog.join()
        current_app.logger.info("TestWithHeartBeat exited")

    def _watch_dog(self):
        self.logger.info('watch_dog started')
        while not self.watch_dog_shutdown_event.is_set():
            self.logger.info(f'watch_dog heartbeat!')
            self.watch_dog_shutdown_event.wait(timeout=3)
        self.logger.info('watch_dog stopped')

I'm trying to run a Celery instance to process some async task, I need a heartbeat thread to check if the task is alive.

starting by command: celery -A cell worker -l debug --pool gevent --concurrency=500

When I use sleep to block the main thread and simulate a time-consuming operation, the heartbeat thread can run normally and print 'watch_dog heartbeat!'. However, when actually executing the time-consuming business logic, the heartbeat thread cannot run, unless I manually insert commands like time.sleep in the code to make the main thread pause.

This is indeed a big problem，If the time-consuming task takes too long to complete and the heartbeat doesn't arrive in time, I'll terminate the task in another service.

How can I ensure that the daemon thread runs normally at the time it is supposed to execute?

本文标签：