My example code is:
- run.py
from celery import Celery, Task
app = Celery(
"app",
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='Asia/Tokyo',
enable_utc=True,
backend='redis://127.0.0.1:6369/2',
broker='redis://127.0.0.1:6369/3',
include=['app.tasks']
)
app.conf.task_routes = {'app.tasks.long_run': {'queue': 'long_running_task'}}
class VerboseTask(Task):
def on_failure(self, exc, task_id, args, kwargs, einfo):
print("-------------------------------------------")
print(
f"FAILURE({task_id}):"
f"{self.request.task}{kwargs}, Exception={exc}, ErrorInfo={einfo}"
)
print("-------------------------------------------")
def on_success(self, retval, task_id, args, kwargs):
print("-------------------------------------------")
print(f"SUCCESS({task_id}):{self.request.task}{kwargs}")
print("-------------------------------------------")
def on_retry(self, exc, task_id, args, kwargs, einfo):
print("-------------------------------------------")
print(f"RETRY({task_id}):{self.request.task}{kwargs}")
print("-------------------------------------------")
- app/tasks.py
from run import app, VerboseTask
from time import sleep
class TSK:
name = 'tsk'
@app.task(bind=True, base=VerboseTask)
def short_run(self, **kwargs):
for i in range(0, 3):
print(f'{self.request.task} [{self.request.id[0:5]}] → {i*5} ~ {(i+1)*5}')
sleep(5)
print('short_run finished')
@app.task(bind=True, base=VerboseTask)
def long_run(self, **kwargs):
for i in range(0, 10):
print(f'{self.request.task} [{self.request.id[0:5]}] → {i*5} ~ {(i+1)*5}')
sleep(5)
print('long_run finished')
And I run celery with
celery -A run worker -l info --concurrency=4 -Q long_running_task,celery
The problem is when I invoke long_running_task four times and then usual celery task, the long_running_task consume all workers and celery task need to wait for four long running tasks.
I want to keep at least 2 workers for short running tasks. For example if current tasks are two short running tasks and one long running task, then only one worker is available and only short running task able run. Long running task have to wait while three workers will be free at same time to consume one of it.
In other way just limiting of max concurrency for long running task to 2 will be also a possible solution.
I know that I may run two Celery instances with separated concurrency:
celery -A run worker -l info --concurrency=2 -n worker1@%h -Q long_running_task
celery -A run worker -l info --concurrency=2 -n worker2@%h -Q celery
But
- I want to consume
long_running_taskworkers for other tasks too if that is idle (that is not so important condition and may be ignored) - I use docker container and want to keep only one
ENTRYPOINTinside it (in other words I don't want to separate workers by running separate commands. I want to executecelery -A run worker ...only once)
Is here any way to customize limits and concurrency for workers that meet my conditions?