通過celery異步處理一個查詢?nèi)蝿?wù)的完整代碼
今天介紹通過celery實現(xiàn)一個異步任務(wù)。有這樣一個需求，前端發(fā)起一個查詢的請求，但是發(fā)起查詢后，查詢可能不會立即返回結(jié)果。這時候，發(fā)起查詢后，后端可以把這次查詢當(dāng)作一個task，并立即返回一個能唯一表明該task的值，如taskID(用戶后面可以通過這個taskID 隨時查看結(jié)果)，用戶收到這個taskID后，可以轉(zhuǎn)去處理其他任務(wù)，而不必一直等待查詢結(jié)果。后端API調(diào)用celery來處理這個task，并將結(jié)果值保存在一個csv文件中，后面用戶通過taskID 查詢時返回結(jié)果。
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參數(shù)只能為type=<?>&projectName=<?>&serviceGroupName=<?>;必須指定三個參數(shù)", {})
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 = "任務(wù)已經(jīng)創(chuàng)建，詳情請查看：http://10.4.34.254/api/task?taskId=%s"% (taskId)
return [resultInfo]
getServiceInfo.apply_async((contentInfo,),queue=”getServiceInfo”),重點是這一行，apply_async()方法會返回一個AsyncResult實例，通過這個實例可以跟蹤任務(wù)狀態(tài)軌跡。
要使用此功能，需要提供結(jié)果后臺(result backend)，這樣才有地方存儲任務(wù)狀態(tài)等信息。其中，getServiceInfo是自定義的一個task，后續(xù)會介紹到，contentInfo是傳遞的一個參數(shù)，queue是指定隊列名稱。
上面這個函數(shù)的原型如下：
task.apply_async(args[, kwargs[, …]])
其中 args 和 kwargs 分別是 task 接收的參數(shù)，當(dāng)然它也接受額外的參數(shù)對任務(wù)進行控制。
在 Celery 中執(zhí)行任務(wù)的方法一共有三種：
1. delay， 用來進行最簡單便捷的任務(wù)執(zhí)行（delay在第3小節(jié)的測試中使用過，它可以看作是apply_async的一個快捷方式）；
2. apply_async， 對于任務(wù)的執(zhí)行附加額外的參數(shù)，對任務(wù)進行控制；
3. app.send_task， 可以執(zhí)行未在 Celery 中進行注冊的任務(wù)。
celery文件配置
在python的庫存放路徑中(一般是/usr/lib/python2.6/site-packages)，創(chuàng)建一個文件夾proj，進入proj目錄，創(chuàng)建三個文件,init,將proj聲明一個python包，celepy,其內(nèi)容如下：
#_*_ 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"
#第三步：連接數(shù)據(jù)庫，執(zhí)行代碼邏輯
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
celery -A proj worker -Q getServiceInfo -l debug -c 6
最后，寫一個結(jié)果，專門獲取查詢結(jié)果的結(jié)果，傳入的參數(shù)為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參數(shù)無比指定taskId這個參數(shù)", {})
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異步處理一個查詢?nèi)蝿?wù)的完整代碼就是小編分享給大家的全部內(nèi)容了，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關(guān)文章
Python?格式化輸出字符串的方法(輸出字符串+數(shù)字的幾種方法)
字符串格式化輸出是python非常重要的基礎(chǔ)語法，這篇文章主要介紹了Python?格式化輸出字符串(輸出字符串+數(shù)字的幾種方法)的方法,需要的朋友可以參考下2023-03-03
今天小編就為大家分享一篇對numpy中shape的深入理解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-06-06
有些時候我們需要每隔一段時間就要執(zhí)行一段程序,或者是往復(fù)循環(huán)執(zhí)行某一個任務(wù),下面這篇文章主要給大家介紹了關(guān)于python定時任務(wù)timeloop庫用法的相關(guān)資料,需要的朋友可以參考下2023-01-01
Python中基本的日期時間處理的學(xué)習(xí)教程
這篇文章主要介紹了Python中基本的日期時間處理的學(xué)習(xí)教程,日期時間相關(guān)模塊的使用是Python入門學(xué)習(xí)中的基礎(chǔ)知識,需要的朋友可以參考下2015-10-10
C++和python實現(xiàn)阿姆斯特朗數(shù)字查找實例代碼
這篇文章主要給大家介紹了關(guān)于C++和python實現(xiàn)阿姆斯特朗數(shù)字查找的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-12-12
python腳本設(shè)置超時機制系統(tǒng)時間的方法
這篇文章主要介紹了python腳本設(shè)置超時機制系統(tǒng)時間的方法,感興趣的小伙伴們可以參考一下2016-02-02
這篇文章主要介紹了Python松散正則表達式用法,較為詳細(xì)的分析了松散正則表達式的概念、功能與相關(guān)使用技巧,需要的朋友可以參考下2016-04-04
最近python課程學(xué)完了，琢磨著用python點什么東西，經(jīng)過一番搜索，盯上了QQ空間，本文主要講述了如何爬取網(wǎng)頁版QQ空間，并生成詞云圖、柱狀圖、折線圖、餅圖的各種示例代碼2021-06-06
Python人工智能深度學(xué)習(xí)RNN模型結(jié)構(gòu)流程
這篇文章主要為大家介紹了Python人工智能深度學(xué)習(xí)RNN的模型流程結(jié)構(gòu)，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步2021-11-11
pytest的主配置文件,可以改變pytest的默認(rèn)行為,本文主要介紹了解析Pytest3種配置文件方式,具有一定的參考價值,感興趣的可以了解一下2024-02-02