在之前的一篇文章中提到了用Django+Celery+Redis實現(xiàn)了異步任務(wù)隊列，只不過消息中間件使用了redis，redis作為消息中間件可謂是差強人意，功能和性能上都不如Rabbitmq，所以本次使用tornado框架結(jié)合celery，同時消息中間件使用Rabbitmq來實現(xiàn)異步發(fā)郵件，并且使用flower來監(jiān)控任務(wù)隊列。

首先安裝rabbitmq

Mac os直接運行brew命令安裝

#安裝服務(wù)

brew install rabbitmq

#啟動服務(wù)

brew services start rabbitmq

Win10系統(tǒng)就要下載安裝包進行安裝了，由于rabbitmq是基于erlang的，所以要首先安裝erlang

1、首先，下載并運行Erlang for Windows 安裝程序 (地址：http://www.erlang.org/downloads)下載完畢并安裝(注意：安裝目錄請選擇默認目錄)

安裝成功后，啟用web管理UI，進入RabbitMQ Serverrabbitmq_server-3.6.6sbin，輸入命令rabbitmq-plugins enable rabbitmq_management

在系統(tǒng)的開始菜單里找到RabbitMQ的啟動菜單，啟動服務(wù)

瀏覽器輸入，http://localhost:15672/，使用默認用戶guest/guest進入網(wǎng)頁端控制臺：

代表沒有問題了

然后安裝tornado和celery，注意指定版本號

pip3 install tornado==5.1.1

pip3 install celery ==3.1

pip3 install pika ==0.9.14

pip3 install tornado-celery

pip3 install flower

需要注意一點，由于python3.7中async已經(jīng)作為關(guān)鍵字存在，但是有的三方庫還沒有及時修正，導(dǎo)致它們自己聲明的變量和系統(tǒng)關(guān)鍵字重名，所以我們要深入三方庫的源碼，幫他們修改async關(guān)鍵字為async_my，需要修改的文件夾和文件包含但不限于：

/site-packages/pika/adapters/libev_connection.py

/site-packages/celery下面的文件

/site-packages/kombu下面的文件夾

在tornado項目下新建一個任務(wù)隊列文件task.py：

import time

from celery import Celery

from func_tool import mail

C_FORCE_ROOT=True

celery = Celery("tasks", broker="amqp://guest:guest@localhost:5672")

celery.conf.CELERY_RESULT_BACKEND = "amqp"

@celery.task

def sleep(seconds):

time.sleep(float(seconds))

return seconds

@celery.task

def sendmail(title,text,tomail):

mail(title,text,tomail)

return '發(fā)送郵件成功'

if __name__ == "__main__":

celery.start()

然后編寫服務(wù)端代碼:

from celery import Celery

from tornado import gen

import tcelery

sys.path.append("..")

import task

#異步任務(wù)

class CeleryHandler(BaseHandler):

@gen.coroutine

def get(self):

response = yield gen.Task(task.sendmail.apply_async,args=['你好','非常好','164850527@qq.com'])

self.write('ok')

self.finish()

路由器代碼:

import tornado.web

from views import Index

import config

#路由

class Application(tornado.web.Application):

def __init__(self):

handlers = [

(r"/celery", Index.CeleryHandler)

]

super(Application,self).__init__(handlers,**config.setting)

程序入口代碼server.py:

import tornado.ioloop

import tornado.httpserver

import config

from application import Application

if __name__ == "__main__":

print('啟動...')

app = Application()

httpServer = tornado.httpserver.HTTPServer(app)

# httpServer.listen(8888)

#綁定端口

httpServer.bind(config.options['port'])

#開啟5個子進程(默認1，若為None或者小于0，開啟對應(yīng)硬件的CPU核心數(shù)個子進程)

httpServer.start(1)

tornado.ioloop.IOLoop.current().start()

進入項目目錄，分別啟動tornado服務(wù)，celery服務(wù)，以及flower服務(wù)

python server.py

celery -A task worker --loglevel=info

celery flower -A task --broker=amqp://guest:guest@localhost:5672//

訪問網(wǎng)址http://localhost:8000/celery 用來觸發(fā)異步任務(wù)

后臺服務(wù)顯示任務(wù)返回值：

進入flower在線任務(wù)監(jiān)控網(wǎng)址：http://localhost:5555/

至此，整個流程就走完了。

總結(jié)

以上是生活随笔為你收集整理的tornado celery mysql_Python3.7+Tornado5+Celery3+Rabbitmq3实现异步队列任务的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。