概述：

????????我們考慮一個(gè)場(chǎng)景，公司有一個(gè)需求，現(xiàn)在需要做一套web系統(tǒng)，而這套系統(tǒng)某些功能需要使用一些開(kāi)源工具的sdk和api，或是運(yùn)行一些耗時(shí)比較大的任務(wù)（單個(gè)大任務(wù)下可能有多個(gè)小任務(wù)），需要一段時(shí)間才能提供執(zhí)行結(jié)果，而前端同事要求不能讓用戶在頁(yè)面等待，需要馬上提供一個(gè)返回結(jié)果給他，任務(wù)執(zhí)行完后可以拿到最終結(jié)果，并且用戶退出web界面或?yàn)g覽器異常關(guān)閉之后，再次返回界面，執(zhí)行的過(guò)程不會(huì)中斷，并且支持多用戶同時(shí)執(zhí)行不同操作的需要。

? ? ? ? 很明顯，這是一個(gè)-異步多線程-的場(chǎng)景，在Python中可以想到的有：

????????1.引入Asyncio模塊，利用多協(xié)程實(shí)現(xiàn)。

????????2.使用Threading模塊，自己編寫(xiě)線程任務(wù)，線程等待，睡眠，釋放線程的過(guò)程。

????????3.使用異步框架，例如Cerely、Tornado、Twisted等等，裝飾異步任務(wù)。

? ? ? ? 這里邊最便捷且開(kāi)發(fā)效率最高的應(yīng)該是使用異步框架，咱們選擇使用Celery來(lái)實(shí)現(xiàn)這個(gè)需求。

Celery介紹：

? ? ? ? 截圖與描述來(lái)自celery官網(wǎng)：Celery - Distributed Task Queue — Celery 5.2.0 documentation

????????Celery 是一個(gè)簡(jiǎn)單、靈活且可靠的分布式系統(tǒng)，用于處理大量消息，同時(shí)為操作提供維護(hù)此類(lèi)系統(tǒng)所需的工具。

????????它是一個(gè)專注于實(shí)時(shí)處理的任務(wù)隊(duì)列，同時(shí)也支持任務(wù)調(diào)度。

????????Celery 擁有龐大而多樣化的用戶和貢獻(xiàn)者社區(qū)，您應(yīng)該加入我們的 IRC?或我們的郵件列表。

????????Celery 是開(kāi)源的，并在BSD 許可下獲得許可。

消費(fèi)者與消費(fèi)結(jié)果：

????????我們除了需要Celery做異步任務(wù)的處理，還需要一個(gè)中間件來(lái)充當(dāng)消費(fèi)者，并保存最終的任務(wù)處理結(jié)果（消費(fèi)結(jié)果），這里有很多中間件可以選，例如常用的消息中間件，rabbitmq，kafka等，還可以使用mysql，redis等作為消費(fèi)者并保存消費(fèi)結(jié)果（因?yàn)樽罱K的處理結(jié)果要返回給前端同事），樓主最終選擇了redis。

Redis安裝與配置：

????????這里不再贅述windows下安裝redis步驟，只介紹linux下安裝redis與配置，我的機(jī)器是centos7.6：

????????yum方式安裝（注意：這樣安裝的redis不是最新版本的，如有對(duì)版本要求比較高的，建議去官網(wǎng)下載源碼包去手動(dòng)安裝，官網(wǎng)地址：Redis，最新版本：6.2.6）

yum -y install redis

????????安裝完成之后配置redis.conf文件：

vi /etc/redis.conf

????????修改這一行，改成 0.0.0.0，這樣別的應(yīng)用和組件才可以訪問(wèn)到redis的服務(wù)與端口:

????????同理，redis的默認(rèn)端口也可以在此配置里修改：

????????還有一些關(guān)閉匿名訪問(wèn)，設(shè)置密碼等配置的修改，項(xiàng)目若要上到公網(wǎng)環(huán)境下，建議配置。

????????啟動(dòng)并測(cè)試redis服務(wù)功能是否正常：

????????啟動(dòng)redis:????????

redis-cli -h 0.0.0.0

????????測(cè)試redis:

1 redis> set name "zzz"2 3 OK4 5 redis> get name6 7 "zzz"

????????記住，代碼并沒(méi)有實(shí)際引用redis，但也需要安裝redis模塊，否則會(huì)報(bào)錯(cuò)。（redis模塊版本不要太高，高了也會(huì)報(bào)錯(cuò)，這些坑都是樓主親自趟過(guò)的，我這里使用2.10.6）

pip install redis==2.10.6

?Celery的安裝和配置：

????????windos和linux下都可以使用pip安裝：

?pip install celery==3.1.25

? ? ? ? 我的項(xiàng)目目錄：（celeryconfig.py與__init__.py文件為celery與redis配置文件）：

??????????

????????在項(xiàng)目中先創(chuàng)建一個(gè)名為config的python目錄，并在__init__.py中導(dǎo)入celery模塊并配置：

__init__.py:

from celery import Celery,platformsplatforms.C_FORCE_ROOT = Trueapp = Celery("prod")  # 創(chuàng)建 Celery 實(shí)例app.config_from_object("kernel.config.celeryconfig")  # 通過(guò) Celery 實(shí)例加載配置模塊

????????platforms.C_FORCE_ROOT = True 這個(gè)配置一定要有，否則會(huì)報(bào)權(quán)限問(wèn)題。

????????在config目錄下的celeryconfig.py中配置任務(wù)隊(duì)列消費(fèi)者與消費(fèi)結(jié)果保存在redis的地址：

celeryconfig.py:

## celery配置BROKER_URL = "redis://redis-host:6379/1"  # 指定 Broker消費(fèi)者，我們使用redis 1號(hào)數(shù)據(jù)庫(kù)CELERY_RESULT_BACKEND = "redis://redis-host:6379/2"  # 指定 Backend，最終消費(fèi)結(jié)果，我們使用redis 2號(hào)數(shù)據(jù)庫(kù)CELERY_TIMEZONE = "Asia/Shanghai"  # 指定時(shí)區(qū)，默認(rèn)是 UTCCELERY_IMPORTS = (  # 指定導(dǎo)入的任務(wù)模塊    "kernel.views.api"   ## 異步任務(wù)代碼文件路徑即可)

????????至此，前期需要的工具準(zhǔn)備工作全部完畢，我們開(kāi)始我們的開(kāi)發(fā)任務(wù)。

異步任務(wù)開(kāi)發(fā)：

? ? ? ? 樓主因?yàn)橹饕?fù)責(zé)后端這塊，這里選擇使用flask來(lái)寫(xiě)，整體的項(xiàng)目模塊與版本，大概羅列下：

????????????????????????Python 3.5.4
????????????????????????Mysql ?5.5.64????????
????????????????????????Celery==3.1.25
????????????????????????Flask==1.1.4
????????????????????????Redis==2.10.6

????????這時(shí)我們與前端同事再次詳細(xì)溝通了下，初步約定如下：

? ? ? ? 1.前端通過(guò)form表單傳數(shù)據(jù)給后端，格式為json，分析：需要解析json數(shù)據(jù)。

????????2.因?yàn)榇嬖陂L(zhǎng)耗時(shí)的任務(wù)，要求一旦前端請(qǐng)求過(guò)來(lái)，后端要馬上返回一個(gè)中間結(jié)果給前端（這樣解決了前端頁(yè)面等待的問(wèn)題），分析：需要馬上提供一個(gè)返回結(jié)果。

? ? ? ? 3.前端最終要拿到任務(wù)的最終執(zhí)行結(jié)果，分析：我們需要把長(zhǎng)耗時(shí)異步任務(wù)的最終結(jié)果推送給前端，需要任務(wù)代碼最后推送執(zhí)行結(jié)果。（自己先定義回調(diào)接口去測(cè)試）

1.后端Flask接口代碼：

文件名稱與路徑：

????????項(xiàng)目名稱-kernel-view-api.py，與celery配置下的任務(wù)模塊對(duì)應(yīng)。

?api.py:

# -*- coding: utf-8 -*-import json, sysimport loggingimport requestsimport datetime,pymysqlimport os,subprocessfrom flask import render_template, Blueprint, request, g, abort, url_for, jsonify, session, redirect,Responsefrom kernel.models.playbook import PlayBook_filefrom kernel.utils import render_response, Retvalfrom kernel.models import dbfrom sqlalchemy import or_,textimport gitlab  ## 導(dǎo)入gitlab模塊from kernel.config import app, cmdb_config,hcacp_configimport pymysql,uuid,hashlib,timefrom datetime import timezonebp = Blueprint("test", __name__)  ## 藍(lán)圖自己定義，這里只是實(shí)例化log = logging.getLogger(__name__)    ## 日志自己定義，這里只是實(shí)例化class status:  ## 定義一些狀態(tài)碼    success = 0    warning = 1    pending = 2    faild = -1## 回調(diào)接口@bp.route("/test/callback/", methods=["GET", "POST"])def ansible_aaa():    data1 = request.get_data(as_text=True)    # data2 = json.loads(data1)    log.info(data1)    return [email protected]("/test/add/", methods=["POST", "GET"])def devops_add():    """        獲取form表單json數(shù)據(jù)    """    # return True    try:        data = request.get_data()        _data = json.loads((str(data, "utf-8")))        print(_data)    except Exception as requestdata_except:        log.error("獲取表單數(shù)據(jù)異常，異常原因：%s" % requestdata_except)        return render_response(status.faild, u"獲取表單數(shù)據(jù)異常，異常原因：%s" % requestdata_except, {})        ## 獲取標(biāo)識(shí)tag的結(jié)果    try:        """        工單json數(shù)據(jù)要帶工單標(biāo)識(shí)符select_tag:        create_project:新建項(xiàng)目申請(qǐng)工單        """        select_tag = _data.get("select_tag")    except Exception as request_select_tag_except:        log.error("獲取表單需求標(biāo)識(shí)select_tag異常，異常原因：%s" % request_select_tag_except)        return render_response(status.faild, u"獲取表單需求標(biāo)識(shí)select_tag異常，異常原因：%s" % request_select_tag_except, {})    try:        """             !--當(dāng)參數(shù)select_tag == create_project 時(shí),建立項(xiàng)目--!         """        if select_tag == "create_project":            projname = _data.get("projname")            add_project_result = add_project.delay(cmdb_config, _data)            return render_response(status.pending, u"devops系統(tǒng)添加項(xiàng)目工單任務(wù)執(zhí)行中--pending--", {"項(xiàng)目中文名稱": projname})    except Exception as do_celery_job_except:        log.error("執(zhí)行異步celery任務(wù)異常，異常原因：%s" % do_celery_job_except)        return render_response(status.faild, u"執(zhí)行異步celery任務(wù)異常，異常原因：%s" % do_celery_job_except, {})

這里代表前端請(qǐng)求過(guò)來(lái)之后，馬上返回一個(gè)執(zhí)行結(jié)果，滿足需求2：

在devops_add接口里執(zhí)行異步任務(wù)：? ? ? ??

????????add_project_result = add_project.delay(cmdb_config, _data)

官網(wǎng)的示例：

? ? ? ? ## 1.擴(kuò)號(hào)里為異步任務(wù)所需的參數(shù)

????????## 2.add_project_result?是異步任務(wù)執(zhí)行的對(duì)象，包含很多屬性方法，下邊介紹一些常用的：

????????獲取任務(wù)結(jié)果和狀態(tài):
????????add_project_result =?task.apply_async()
????????add_project_result.ready()?????#?查看任務(wù)狀態(tài)，返回布爾值,??任務(wù)執(zhí)行完成,?返回?True,?否則返回?False.
????????add_project_result.wait()??????#?會(huì)阻塞等待任務(wù)完成,?返回任務(wù)執(zhí)行結(jié)果，很少使用；
????????add_project_result.get(timeout=1)???????#?獲取任務(wù)執(zhí)行結(jié)果，可以設(shè)置等待時(shí)間，如果超時(shí)但任務(wù)未完成返回None；
????????add_project_result.result??????#?任務(wù)執(zhí)行結(jié)果，未完成返回None；
????????add_project_result.state???????#?PENDING,?START,?SUCCESS，任務(wù)當(dāng)前的狀態(tài)
????????add_project_result.status??????#?PENDING,?START,?SUCCESS，任務(wù)當(dāng)前的狀態(tài)
????????add_project_result.successful??#?任務(wù)成功返回true
????????add_project_result.traceback??#?如果任務(wù)拋出了一個(gè)異常，可以獲取原始的回溯信息

? ? ?

2.異步任務(wù)代碼：

文件名稱與路徑：

????????項(xiàng)目名稱-kernel-view-api.py

api.py

解釋：

????????因?yàn)橐獫M足需求3，把最終異步耗時(shí)任務(wù)的真正結(jié)果給到前端，所以我們需要在異步任務(wù)里寫(xiě)一個(gè)回調(diào)的操作。

?????????header = {"Content-Type": "application/json"}? ## 構(gòu)造請(qǐng)求頭和數(shù)據(jù)類(lèi)型
????????_json = {"status": sttaus.faild, "msg": u"失敗", "data": {}}? ## 失敗就返回給前端json類(lèi)型失敗

????????_json = {"status": sttaus.success, "msg": u"成功", "data": {}}? ## 成功就返回給前端json類(lèi)型成功

????????requests.post(callback_url, headers=header, data=json.dumps(_json)) ## 帶參回調(diào)請(qǐng)求

# -*- coding: utf-8 -*-import json, sysimport loggingimport requestsimport datetime,pymysqlimport os,subprocessfrom flask import render_template, Blueprint, request, g, abort, url_for, jsonify, session, redirect,Responsefrom kernel.utils import render_response, Retvalfrom datetime import timezonefrom kernel.config import *  ## 導(dǎo)入config目錄下的celery配置bp = Blueprint("test", __name__)  ## 藍(lán)圖自己定義，這里只是實(shí)例化log = logging.getLogger(__name__)    ## 日志自己定義，這里只是實(shí)例化class status:  ## 定義一些狀態(tài)碼    success = 0    warning = 1    pending = 2    faild = -1## 示例函數(shù)：一個(gè)添加信息函數(shù)，前端給我們json數(shù)據(jù)，后端接受之后去插入數(shù)據(jù)庫(kù)，完成操作并告訴前端@app.task  ## celery添加項(xiàng)目任務(wù)def add_project(mysql_config, _data):    try:        ## 系統(tǒng)添加項(xiàng)目信息工單        projname = _data.get("projname")  ## 項(xiàng)目名稱，必填        prodesc= _data.get("prodesc")  ## 項(xiàng)目描述，必填        projctime = datetime.datetime.now()  ## 項(xiàng)目發(fā)布時(shí)間        callback_url = _data.get("callback_url")  ## 回調(diào)接口地址    except Exception as describe_form_except:        log.error("解析表單數(shù)據(jù)出現(xiàn)異常，異常原因：%s" % describe_form_except)        header = {"Content-Type": "application/json"}  ## 回調(diào)接口請(qǐng)求頭        _json = {"status": status.faild, "msg": u"失敗", "data": {}}        requests.post(callback_url, headers=header, data=json.dumps(_json))    try:        # 獲取數(shù)據(jù)庫(kù)連接        conn = pymysql.connect(cmdb_config.server, cmdb_config.user, cmdb_config.password, database=cmdb_config.db)        # 返回連接        cursor = conn.cursor()    except Exception as connect_except:        log.error("系統(tǒng)數(shù)據(jù)庫(kù)連接出現(xiàn)異常，異常原因：%s" % connect_except)        _json = {"status": status.faild, "msg": u"失敗", "data": {}}        requests.post(callback_url, headers=header, data=json.dumps(_json))    try:        proj_sql = "insert into project_tb_project (projname,prodesc,projctime) VALUES ("{}","{}","{}");".format(projname, prodesc, projctime)        cursor.execute(proj_sql)        conn.commit()        _json = {"status": status.success, "msg": u"成功", "data": {}}        requests.post(callback_url, headers=header, data=json.dumps(_json))        ## 任務(wù)執(zhí)行完成之后調(diào)用回調(diào)接口，返回任務(wù)執(zhí)行成功結(jié)果        log.info("系統(tǒng)建項(xiàng)目工單執(zhí)行成功,%s" % proj_sql)    except Exception as do_add_project_except:        _json = {"status": status.faild, "msg": u"失敗", "data": {}}        requests.post(callback_url, headers=header, data=json.dumps(_json))        log.error("執(zhí)行添加項(xiàng)目工單異常，異常原因：%s" % do_add_project_except)        ## 任務(wù)執(zhí)行完成之后調(diào)用回調(diào)接口，返回任務(wù)執(zhí)行失敗結(jié)果

????????樓主用的最簡(jiǎn)單，沒(méi)有在task里寫(xiě)一些屬性，類(lèi)似下邊的這種方式還可以給task添加一些屬性：

[email protected](name="test",bind=True,base=BaseTask)

? ? ? ?補(bǔ)充介紹下異步task有的一些屬性：

????????TASK的一般屬性:
????????Task.name:任務(wù)名稱；
????????Task.request：當(dāng)前任務(wù)的信息；
????????Task.max_retries：設(shè)置重試的最大次數(shù)
????????Task.throws：預(yù)期錯(cuò)誤類(lèi)的可選元組，不應(yīng)被視為實(shí)際錯(cuò)誤，而是結(jié)果失?。?br /> ????????Task.rate_limit：設(shè)置此任務(wù)類(lèi)型的速率限制
????????Task.time_limit：此任務(wù)的硬限時(shí)（以秒為單位）。
????????Task.ignore_result：不存儲(chǔ)任務(wù)狀態(tài)。默認(rèn)False；
????????Task.store_errors_even_if_ignored：如果True，即使任務(wù)配置為忽略結(jié)果，也會(huì)存儲(chǔ)錯(cuò)誤。
????????Task.serializer：標(biāo)識(shí)要使用的默認(rèn)序列化方法的字符串。
????????Task.compression：標(biāo)識(shí)要使用的默認(rèn)壓縮方案的字符串。默認(rèn)為task_compression設(shè)置。
????????Task.backend：指定該任務(wù)的結(jié)果存儲(chǔ)后端用于此任務(wù)。
????????Task.acks_late：如果設(shè)置True為此任務(wù)的消息將在任務(wù)執(zhí)行后確認(rèn)?，而不是在執(zhí)行任務(wù)之前（默認(rèn)行為），即默認(rèn)任務(wù)執(zhí)行之前就會(huì)發(fā)送確認(rèn)；
????????Task.track_started：如果True任務(wù)在工作人員執(zhí)行任務(wù)時(shí)將其狀態(tài)報(bào)告為“已啟動(dòng)”。默認(rèn)是False；

我們啟動(dòng)celery來(lái)看下celery里在執(zhí)行任務(wù)的過(guò)程中有什么變化：

（1）啟動(dòng)項(xiàng)目：

樓主用的是gunicorn工具啟動(dòng)，配置多線程：

gunicorn.conf

????????workers = 16? ?## 多線程配置

????????bind = "0.0.0.0:7777"

????????proc_name = "websocket（項(xiàng)目名稱）"

????????limit_request_field_size = 0

????????limit_request_line = 0

????????log_level = "error"

????????debug = True

????????chdir = "/data/websocket" ## 項(xiàng)目目錄

????????啟動(dòng)命令：gunicorn -c? /項(xiàng)目目錄/gunicorn.conf kernel:app

（2）啟動(dòng)celery：

????????cd 到項(xiàng)目目錄下，執(zhí)行 celery -A kernel.views.api worker -l info ?

（3）使用postman調(diào)用接口：

????????可以看到直接先返回我們狀態(tài)碼2-等待狀態(tài)：

（4）從日志看異步任務(wù)執(zhí)行過(guò)程：

????????1.會(huì)先在celery里出現(xiàn)一個(gè)異步任務(wù)，并生成一個(gè)異步任務(wù)的task-id號(hào)：

????????2.redis去查看是否已有task任務(wù)，task-id號(hào)是一致的：

????????用add_project_result保存異步任務(wù)執(zhí)行結(jié)果的對(duì)象，最終的結(jié)果是在redis中，我們也可以去redis里去拿，redis保存的結(jié)果。

????????我們用的redis 2號(hào)數(shù)據(jù)庫(kù)，select 2 號(hào)數(shù)據(jù)庫(kù)，keys * 查看redis是否已有任務(wù)

????????任務(wù)最終的執(zhí)行結(jié)果（celery日志里也可以看到，在redis里也可以看到，celery日志看的更直觀，succeded代表異步任務(wù)執(zhí)行成功）：

????????3. 查看項(xiàng)目日志，狀態(tài)碼為1，是回調(diào)接口打印出來(lái)的，代表返回給回調(diào)接口最終結(jié)果是成功。

? ? ? ? 4.最終去數(shù)據(jù)庫(kù)看下新添加記錄是否已有，這里就不截圖了，記錄插入成功，異步任務(wù)執(zhí)行成功，也滿足了開(kāi)始我們溝通的三個(gè)需求。

? ? ? ? 5.前端同學(xué)給你豎起了大拇指，直呼你牛！

? ? ? ? ??

?

備注：

??????????????????????????????????????????????????????????????????????????????????????????????????

?

?

? ? ? ? celery還可以用來(lái)做定時(shí)任務(wù)，感興趣的伙伴們可以去官網(wǎng)或者其他途徑去研究下，樓主第一次寫(xiě)這么大的博客，有些地方我描述不清楚的或者您沒(méi)太看懂的可以私信我答疑解惑，我的微信zcw576020095,熱愛(ài)python，熱愛(ài)運(yùn)維，一起加油！

????????