The following script, run as part of the deployment solved the problem:
import time
from celery.app.control import Control
from myapp.tasks import celery # my application's Celery app
if __name__ == "__main__":
control = Control(celery)
control.cancel_consumer("celery") # queue name, must probably be specified once per queue, but my app uses a single queue
inspect = control.inspect()
while True:
active = inspect.active()
running_jobs = []
for key, value in active.items():
running_jobs.extend(value)
if len(running_jobs) > 0:
print("{} jobs running: {}".format(len(running_jobs), ", ".join(job["name"] for job in running_jobs)))
time.sleep(10)
else:
print("No running jobs")
break
与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…