利用Celery实现异步查询任务：从发起请求到获取结果的完整流程
本文介绍如何使用Python的Celery库处理前端发起的长时间查询请求。通过将查询封装为异步任务，后端立即返回TaskID，用户可后续通过该ID查询存储在CSV文件中的结果。文章涵盖apply_async用法、三种任务执行方式、Celery配置及Worker启动命令，提供完整代码示例。
Celery异步任务处理概述
本文介绍如何利用Celery实现异步任务的完整流程。假设前端发起一个查询请求，但该查询可能不会立即返回结果。此时，后端可以将此次查询视为一个独立任务（task），并立即返回一个能唯一标识该任务的值，例如TaskID。用户拿到TaskID后，可以转而处理其他业务，无需阻塞等待查询结果。后端API调用Celery执行该任务，并将最终结果保存至CSV文件，用户后续通过TaskID即可查询并获取结果。
def application(environ,start_response):
"""部分代码省略"""
query_string = environ['QUERY_STRING']
serviceGroupName = ""
for getParam in query_string.split("&"):
params = getParam.split("=")
resultInfo = ""
if params[0] == "type":
alertType = params[1]
elif params[0] == "projectName":
projectName = params[1]
elif params[0] == "serviceGroupName":
serviceGroupName = params[1]
else:
resultInfo = error_info(-1, "GET参数只能为type=<?>&projectName=<?>&serviceGroupName=<?>;必须指定三个参数", {})
return [resultInfo]
taskId = 1
result_file_name = '/var/www/dba_api/api/test/'+ str(taskId) + '.csv'
contentInfo = json.dumps({"taskId":1,"opType":"continue","serviceGroupName":serviceGroupName,"dbHost":dbHost,"dbPasswd":dbPasswd,"dbUser":dbUser,"dbPort":dbPort})
result = getServiceInfo.apply_async((contentInfo,),queue="getServiceInfo")
taskInfo = "任务已经创建，详情请查看：http://10.4.34.254/api/task?taskId=%s"% (taskId)
return [resultInfo]核心方法：apply_async与任务追踪
核心代码为 `getServiceInfo.apply_async((contentInfo,), queue="getServiceInfo")`。关键在于 `apply_async()` 方法会返回一个 `AsyncResult` 实例，通过该实例可以实时跟踪任务的执行状态和轨迹。
要使用此功能，必须配置结果后端（result backend），以便存储任务状态等关键信息。其中，`getServiceInfo` 是自定义的任务函数（后续章节详细介绍），`contentInfo` 是传递给任务的具体参数，而 `queue` 参数用于指定任务所属的队列名称。
上述函数的原型定义如下：
`task.apply_async(args[, kwargs[, …]])`
其中，`args` 和 `kwargs` 分别代表任务接收的位置参数和关键字参数。此外，该方法还支持传入额外的控制参数，用于精细调控任务的执行行为。
Celery任务的三种执行方式
在Celery中，执行任务主要有以下三种方式：
1. `delay`：用于执行最简单、便捷的任务。`delay` 在第3小节的测试中已演示，它实际上可以看作是 `apply_async` 的一种快捷封装方式；
2. `apply_async`：适用于需要附加额外参数以控制任务执行场景的情况；
3. `app.send_task`：用于执行那些未在Celery中预先注册的任务。
Celery配置文件详解
接下来介绍Celery文件的配置方法。
在Python库存放路径（通常为 `/usr/lib/python2.6/site-packages`）下，创建一个名为 `proj` 的文件夹。进入 `proj` 目录后，创建三个文件：`__init__.py`（将 `proj` 声明为Python包）、`celery.py`（内容如下）：
#_*_ coding:utf-8 _*_
from __future__ import absolute_import
from celery import Celery
app = Celery("proj",
broker="amqp://user:password@localhost//",
backend="amqp",
include=["proj.tasks"]
)
app.conf.update(
CELERY_ROUTES={
"proj.tasks.getServerInfo":{"queue":"getServerInfo"},
}
)
if __name__=="__main__":
app.start()
此处定义了模块名称 `proj` 以及Celery的路由规则。
此外，还需要创建另一个文件 `task.py`。
#_*_ coding:utf-8 _*_i
from __future__ import absolute_import
from proj.celery import app
import random
import simplejson as json
import types
import time
import MySQLdb
import urllib2
import ConfigParser as cparser
import hmac
import hashlib
import base64
@app.task
def getServiceInfo(contentInfo):
contentInfo = json.loads(contentInfo)
serviceGroupName = contentInfo['serviceGroupName']
dbHost = contentInfo['dbHost']
dbPort = int(contentInfo['dbPort'])
dbUser = contentInfo['dbUser']
dbPasswd = contentInfo['dbPasswd']
msgLib = MessageLib.MessageLib()
Sql = "Your SQL"
#第三步：连接数据库，执行代码逻辑
try:
db_connection = MySQLdb.connect(host=dbHost, port=dbPort, passwd=dbPasswd, db="cmdb", user=dbUser, connect_timeout=2, charset="utf8")
cursor = db_connection.cursor()
cursor.execute(getServiceGroupHostSql)
row = cursor.fetchall()
result = []
for line in row:
...
result.append(tempMysqlHighInfo)
resultInfo = msgLib.success_info(result)
return resultInfo
except Exception, e:
raise
errorInfo = "dbhost:%s, port:%s, error:%s" % (dbHost, dbPort, str(e))
#return getServiceGroupHostSql,errorInfo
return msgLib.error_info(-1, errorInfo, {})
启动Celery Worker与结果获取
最后，介绍如何启动Celery Worker。
启动命令为：`celery -A proj worker -Q getServiceInfo -l debug -c 6`
最后，编写一个专门用于获取查询结果的接口，传入参数为TaskID。部分核心代码如下：
def application(environ,start_response):
status = '400 ERROR'
response_headers = [('Content-type', 'application/json;charset=utf-8')]
start_response(status, response_headers)
status = '200 OK'
response_headers = [('Content-type', 'application/json;charset=utf-8')]
start_response(status, response_headers)
if environ['REQUEST_METHOD'] != "GET":
resultInfo = msgLib.error_info(-1, "http请求类型不是GET", {})
return [resultInfo]
query_string = environ['QUERY_STRING']
serviceGroupName = ""
for getParam in query_string.split("&"):
params = getParam.split("=")
resultInfo = ""
if params[0] == "taskId":
taskId = params[1]
else:
resultInfo = msgLib.error_info(-1, "GET参数无比指定taskId这个参数", {})
return [resultInfo]
logging.info(query_string)
result_file_name = '/var/www/dba_api/api/test/'+ str(taskId) + '.csv'
result = []
try:
with open (result_file_name,'rb') as fp:
lines = csv.reader(fp)
for line in lines :
result.append(line)
resultInfo = msgLib.success_info(result)
return resultInfo
except Exception, e:
errorInfo = "some thing wrong"
return msgLib.error_info(-1, errorInfo, {})
以上即为通过Celery异步处理查询任务的完整代码示例，供读者参考与实践。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
