This is how I set up celery with django on elastic beanstalk with scalability working fine. Please keep in mind that ‘leader_only’ option for container_commands works only on environment rebuild or deployment of the App. If service works long enough, leader node may be removed by Elastic Beanstalk. To deal with that, you may have … Read more
It depends what you want your tasks to be doing, if you need to distribute them, and how you want to manage them. A crontab is capable of executing a script every N intervals. It runs, and then returns. Essentially you get a single execution each interval. You could just direct a crontab to execute … Read more
Use YourTask.apply_async(args=[some, args, here], eta=when) And at the end of your task, reschedule it to the next time it should run.
You can get this kind of behavior using an intermediate task. Here’s a demonstration of creating a “map” like method that works like you’ve suggested. from celery import task, subtask, group @task def get_list(amount): return [i for i in range(amount)] @task def process_item(item): # do stuff pass @task def dmap(it, callback): # Map a callback … Read more
I think you need to restart the worker server. I meet the same problem and solve it by restarting.
In order for Celery to know what the current state of the task is, it sets some metadata in whatever result backend you have. You can piggy-back on that to store other kinds of metadata. def yielder(): for i in range(2**100): yield i @task def report_progress(): for progress in yielder(): # set current progress on … Read more
Based on MattH’s answer, you could use a decorator like this: def single_instance_task(timeout): def task_exc(func): @functools.wraps(func) def wrapper(*args, **kwargs): lock_id = “celery-single-instance-” + func.__name__ acquire_lock = lambda: cache.add(lock_id, “true”, timeout) release_lock = lambda: cache.delete(lock_id) if acquire_lock(): try: func(*args, **kwargs) finally: release_lock() return wrapper return task_exc then, use it like so… @periodic_task(run_every=timedelta(minutes=1)) @single_instance_task(60*10) def fetch_articles() yada … Read more
The twisted reactor cannot be restarted. A work around for this is to let the celery task fork a new child process for each crawl you want to execute as proposed in the following post: Running Scrapy spiders in a Celery task This gets around the “reactor cannot be restart-able” issue by utilizing the multiprocessing … Read more
Return the task_id (which is given from .delay()) and ask the celery instance afterwards about the state: x = method.delay(1,2) print x.task_id When asking, get a new AsyncResult using this task_id: from celery.result import AsyncResult res = AsyncResult(“your-task-id”) res.ready()
Celery 4.0+ does not officially support window already. But it still works on window for some development/test purpose. Use eventlet instead as below: pip install eventlet celery -A <module> worker -l info -P eventlet It works for me on window 10 + celery 4.1 + python 3. This solution solved the following exception: [2017-11-16 21:19:46,938: … Read more