利用Celery实现异步查询任务：从发起请求到获取结果的完整流程

一只小鹿

发布时间：2026-10-13 | 3人浏览过

原创

本文介绍如何使用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异步处理查询任务的完整代码示例，供读者参考与实践。