标签：
If you‘ve worked with Django at some point you probably had the need for some background processing of long running tasks. Chances are you‘ve used some sort of task queue, and Celery is currently the most popular project for this sort of thing in the Python (and Django) world (but there are others).
While working on some projects that used Celery for a task queue I‘ve gathered a number of best practices and decided to document them. Nevertheless, this is more a rant about what I think should be the proper way to do things, and about some underused features that the celery ecosystem offers.
Let me explain why I think this is wrong (aside from the limitations pointed out in the celery docs).
A database is not built for doing the things a proper AMQP broker like RabbitMQ is designed for. It will break down at one point, probably in production with not that much traffic/user base.
I guess the most popular reason people decide to use a database is because, well, they already have one for their web app, so why not re-use it. Setting up is a breeze and you don‘t need to worry about another component (like RabbitMQ).
Not so hypothetical scenario: Let‘s say you have 4 background workers processing the tasks you‘ve put in the database. This means that you get 4 processes polling the database for new tasks fairly often, not to mention that each of those 4 workers can have multiple concurrent threads of it‘s own. At some point you notice that you are falling behind on your task processing and more tasks are coming in than are being completed, so naturally you increase the number of workers doing the task processing. Suddenly your database starts falling apart due to the huge number of workers polling the database for new tasks, your disk IO goes through the roof and your webapp starts being affected by this slow down because the workers are basically DDOS-ing the database.
This does not happen when you have a proper AMQP like RabbitMQ because, for one thing, the queue resides in memory so you don‘t hammer your disk. The consumers (the workers) do not need to resort to polling as the queue has a way of pushing new tasks to the consumers, and if the AMQP does get overwhelmed for some other reason, at least it will not bring down the user facing web app with it.
I would go as far to say that you shouldn‘t use a database for a broker even in development, what with things like Docker and a ton of pre-built images that already give you RabbitMQ out of the box.
Celery is fairly simple to set up, and it comes with a default queue in which it puts all the tasks unless you tell it otherwise. The most common thing you‘ll see is something like this:
@app.task()
def my_taskA(a, b, c):
print("doing something here...")
@app.task()
def my_taskB(x, y):
print("doing something here...")
What happens here is that both tasks will end up in the same Queue (if not specified otherwise in the celeryconfig.py file). I can definitely see the appeal of doing something like this because with just one decorator you‘ve got yourself some sweet background tasks. My concern here is that taskA and taskB might be doing totally different things, and perhaps one of them might even be much more important than the other, so why throw them both in the same basket? Even if you‘ve got just one worker processing both tasks, suppose that at some point the unimportant taskB gets so massive in numbers that the more important taksA just can‘t get enough attention from the worker? At this point increasing the number of workers will probably not solve your problem as all workers still need to process both tasks, and with taskB so great in numbers taskA still can‘t get the attention it deserves. Which brings us to the next point.
The way to solve the issue above is to have taskA in one queue, and taskB in another and then assign x workers to process Q1 and all the other workers to process the more intensive Q2 as it has more tasks coming in. This way you can still make sure that taskB gets enough workers all the while maintaining a few priority workers that just need to process taskA when one comes in without making it wait to long on processing.
So, define your queues manually:
CELERY_QUEUES = (
Queue(‘default‘, Exchange(‘default‘), routing_key=‘default‘),
Queue(‘for_task_A‘, Exchange(‘for_task_A‘), routing_key=‘for_task_A‘),
Queue(‘for_task_B‘, Exchange(‘for_task_B‘), routing_key=‘for_task_B‘),
)
And your routes that will decide which task goes where:
CELERY_ROUTES = {
‘my_taskA‘: {‘queue‘: ‘for_task_A‘, ‘routing_key‘: ‘for_task_A‘},
‘my_taskB‘: {‘queue‘: ‘for_task_B‘, ‘routing_key‘: ‘for_task_B‘},
}
Which will allow you to run workers for each task:
celery worker -E -l INFO -n workerA -Q for_task_A celery worker -E -l INFO -n workerB -Q for_task_B
Most tasks I‘ve seen in the wild don‘t have a notion of error handling at all. If a task fails that‘s it, it failed. This might be fine for some use cases, however, most tasks I‘ve seen are talking to some kind of 3rd party API and fail because of some sort of network error, or other kind of "resource availability" error. The most simple way we can handle these kinds of errors is to just retry the task, because maybe the 3rd party API just had some server/network issues and it will be back up shortly, why not give it a go?
@app.task(bind=True, default_retry_delay=300, max_retries=5)
def my_task_A():
try:
print("doing stuff here...")
except SomeNetworkException as e:
print("maybe do some clenup here....")
self.retry(e)
What I like to do is define per task defaults for how long should a task wait before being retried, and how many retries is enough before finally giving up (the default_retry_delay and max_retries parameters respectively). This is the most basic form of error handling that I can think of and yet I see it used almost never. Of course Celery offers more in terms of error handling but I‘ll leave you with the celery docs for that.
The Flower project is a wonderful tool for monitoring your celery tasks and workers. It‘s web based and allows you to do stuff like see task progress, details, worker status, bringing up new workers and so forth. Check out the full list of features in the provided link.
A task status is the information about the task exiting with a success or failure. It can be useful for some kind of statistics later on. The big thing to note here is that the exit status is not the result of the job that the task was performing, that information is most likely some sort of side effect that gets written to the database (ie. update a user‘s friend list).
Most projects I‘ve seen don‘t really care about keeping persistent track of a task‘s status after it exited yet most of them use either the default sqlite database for saving this information, or even better, they‘ve taken the time and use their regular database (postgres or otherwise).
Why hammer your webapp‘s database for no reason? Use CELERY_IGNORE_RESULT = True in your celeryconfig.py and discard the results.
After giving this talk at a local Python meetup a few people suggested I add this to the list. What‘s it all about? You shouldn‘t pass Database objects (for instance your User model) to a background task because the serialized object might contain stale data. What you want to do is feed the task the User id and have the task ask the database for a fresh User object.
标签：
原文地址：http://www.cnblogs.com/UnGeek/p/5908244.html
