python中的非同步任務隊列（python 循環隊列）

本文目錄一覽：

• 1、如何用python簡單的設計開發非同步任務調度隊列
• 2、python2.7怎麼實現非同步
• 3、Python實現簡單多線程任務隊列
• 4、Python 非同步任務隊列Celery 使用
• 5、python實現堆棧與隊列的方法
• 6、python 里有沒有輕量的持久化隊列

如何用python簡單的設計開發非同步任務調度隊列

經常有人在群里問，運維人員需不需要學開發？需不需要學 PYTHON ？ PYTHON 和 SHELL 有什麼區別？天天問這種好水的問題，我實在受不了，決定幫大家掃掃盲，求求新手們，以後別他媽瞎問了。 現階段，掌握一門開發語言已經成為高級運維工程師的必…

python2.7怎麼實現非同步

改進之前

之前，我的查詢步驟很簡單，就是：

前端提交查詢請求 — 建立資料庫連接 — 新建游標 — 執行命令 — 接受結果 — 關閉游標、連接

這幾大步驟的順序執行。

這裡面當然問題很大：

建立資料庫連接實際上就是新建一個套接字。這是進程間通信的幾種方法里，開銷最大的了。

在「執行命令」和「接受結果」兩個步驟中，線程在阻塞在資料庫內部的運行過程中，資料庫連接和游標都處於閑置狀態。

這樣一來，每一次查詢都要順序的新建資料庫連接，都要阻塞在資料庫返回結果的過程中。當前端提交大量查詢請求時，查詢效率肯定是很低的。

第一次改進

之前的模塊里，問題最大的就是第一步——建立資料庫連接套接字了。如果能夠一次性建立連接，之後查詢能夠反覆服用這個連接就好了。

所以，首先應該把資料庫查詢模塊作為一個單獨的守護進程去執行，而前端app作為主進程響應用戶的點擊操作。那麼兩條進程怎麼傳遞消息呢？翻了幾天Python文檔，終於構思出來：用隊列queue作為生產者（web前端）向消費者（資料庫後端）傳遞任務的渠道。生產者，會與SQL命令一起，同時傳遞一個管道pipe的連接對象，作為任務完成後，回傳結果的渠道。確保，任務的接收方與發送方保持一致。

作為第二個問題的解決方法，可以使用線程池來並發獲取任務隊列中的task，然後執行命令並回傳結果。

第二次改進

第一次改進的效果還是很明顯的，不用任何測試手段。直接點擊頁面鏈接，可以很直觀地感覺到反應速度有很明顯的加快。

但是對於第二個問題，使用線程池還是有些欠妥當。因為，CPython解釋器存在GIL問題，所有線程實際上都在一個解釋器進程里調度。線程稍微開多一點，解釋器進程就會頻繁的切換線程，而線程切換的開銷也不小。線程多一點，甚至會出現「抖動」問題（也就是剛剛喚醒一個線程，就進入掛起狀態，剛剛換到棧幀或內存的上下文，又被換回內存或者磁碟），效率大大降低。也就是說，線程池的並發量很有限。

試過了多進程、多線程，只能在單個線程里做文章了。

Python中的asyncio庫

Python里有大量的協程庫可以實現單線程內的並發操作，比如Twisted、Gevent等等。Python官方在3.5版本里提供了asyncio庫同樣可以實現協程並發。asyncio庫大大降低了Python中協程的實現難度，就像定義普通函數那樣就可以了，只是要在def前面多加一個async關鍵詞。async def函數中，需要阻塞在其他async def函數的位置前面可以加上await關鍵詞。

import asyncio

async def wait():

await asyncio.sleep(2)

async def execute(task):

process_task(task)

await wait()

continue_job()

async def函數的執行稍微麻煩點。需要首先獲取一個loop對象，然後由這個對象代為執行async def函數。

loop = asyncio.get_event_loop()

loop.run_until_complete(execute(task))

loop.close()

loop在執行execute(task)函數時，如果遇到await關鍵字，就會暫時掛起當前協程，轉而去執行其他阻塞在await關鍵詞的協程，從而實現協程並發。

不過需要注意的是，run_until_complete()函數本身是一個阻塞函數。也就是說，當前線程會等候一個run_until_complete()函數執行完畢之後，才會繼續執行下一部函數。所以下面這段代碼並不能並發執行。

for task in task_list:

loop.run_until_complete(task)

對與這個問題，asyncio庫也有相應的解決方案：gather函數。

loop = asyncio.get_event_loop()

tasks = [asyncio.ensure_future(execute(task))

for task in task_list]

loop.run_until_complete(asyncio.gather(*tasks))

loop.close()

當然了，async def函數的執行並不只有這兩種解決方案，還有call_soon與run_forever的配合執行等等，更多內容還請參考官方文檔。

Python下的I/O多路復用

協程，實際上，也存在上下文切換，只不過開銷很輕微。而I/O多路復用則完全不存在這個問題。

目前，Linux上比較火的I/O多路復用API要算epoll了。Tornado，就是通過調用C語言封裝的epoll庫，成功解決了C10K問題（當然還有Pypy的功勞）。

在Linux里查文檔，可以看到epoll只有三類函數，調用起來比較方便易懂。

創建epoll對象，並返回其對應的文件描述符（file descriptor）。

int epoll_create(int size);

int epoll_create1(int flags);

控制監聽事件。第一個參數epfd就對應於前面命令創建的epoll對象的文件描述符；第二個參數表示該命令要執行的動作：監聽事件的新增、修改或者刪除；第三個參數，是要監聽的文件對應的描述符；第四個，代表要監聽的事件。

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

等候。這是一個阻塞函數，調用者會等候內核通知所註冊的事件被觸發。

int epoll_wait(int epfd, struct epoll_event *events,

int maxevents, int timeout);

int epoll_pwait(int epfd, struct epoll_event *events,

int maxevents, int timeout,

const sigset_t *sigmask);

在Python的select庫里：

select.epoll()對應於第一類創建函數；

epoll.register()，epoll.unregister()，epoll.modify()均是對控制函數epoll_ctl的封裝；

epoll.poll()則是對等候函數epoll_wait的封裝。

Python里epoll相關API的最大問題應該是在epoll.poll()。相比於其所封裝的epoll_wait，用戶無法手動指定要等候的事件，也就是後者的第二個參數struct epoll_event *events。沒法實現精確控制。因此只能使用替代方案：select.select()函數。

根據Python官方文檔，select.select(rlist, wlist, xlist[, timeout])是對Unix系統中select函數的直接調用，與C語言API的傳參很接近。前三個參數都是列表，其中的元素都是要註冊到內核的文件描述符。如果想用自定義類，就要確保實現了fileno()方法。

其分別對應於：

rlist: 等候直到可讀

wlist: 等候直到可寫

xlist: 等候直到異常。這個異常的定義，要查看系統文檔。

select.select()，類似於epoll.poll()，先註冊文件和事件，然後保持等候內核通知，是阻塞函數。

實際應用

Psycopg2庫支持對非同步和協程，但和一般情況下的用法略有區別。普通資料庫連接支持不同線程中的不同游標並發查詢；而非同步連接則不支持不同游標的同時查詢。所以非同步連接的不同游標之間必須使用I/O復用方法來協調調度。

所以，我的大致實現思路是這樣的：首先並發執行大量協程，從任務隊列中提取任務，再向連接池請求連接，創建游標，然後執行命令，並返回結果。在獲取游標和接受查詢結果之前，均要阻塞等候內核通知連接可用。

其中，連接池返回連接時，會根據引用連接的協程數量，返回負載最輕的連接。這也是自己定義AsyncConnectionPool類的目的。

我的代碼位於：bottle-blog/dbservice.py

存在問題

當然了，這個流程目前還一些問題。

首先就是每次輪詢拿到任務之後，都會走這麼一個流程。

獲取連接 — 新建游標 — 執行任務 — 關閉游標 — 取消連接引用

本來，最好的情況應該是：在輪詢之前，就建好游標；在輪詢時，直接等候內核通知，執行相應任務。這樣可以減少輪詢時的任務量。但是如果協程提前對應好連接，那就不能保證在獲取任務時，保持各連接負載均衡了。

所以這一塊，還有工作要做。

還有就是epoll沒能用上，有些遺憾。

以後打算寫點C語言的內容，或者用Python/C API，或者用Ctypes包裝共享庫，來實現epoll的調用。

最後，請允許我吐槽一下Python的epoll相關文檔：簡直太弱了！！！必須看源碼才能弄清楚功能。

Python實現簡單多線程任務隊列

Python實現簡單多線程任務隊列

最近我在用梯度下降演算法繪製神經網路的數據時，遇到了一些演算法性能的問題。梯度下降演算法的代碼如下（偽代碼）：

defgradient_descent(): # the gradient descent code plotly.write(X, Y)

一般來說，當網路請求 plot.ly 繪圖時會阻塞等待返回，於是也會影響到其他的梯度下降函數的執行速度。

一種解決辦法是每調用一次 plotly.write 函數就開啟一個新的線程，但是這種方法感覺不是很好。 我不想用一個像 cerely（一種分散式任務隊列）一樣大而全的任務隊列框架，因為框架對於我的這點需求來說太重了，並且我的繪圖也並不需要 redis 來持久化數據。

那用什麼辦法解決呢？我在 python 中寫了一個很小的任務隊列，它可以在一個單獨的線程中調用 plotly.write函數。下面是程序代碼。

fromthreadingimportThreadimportQueueimporttime classTaskQueue(Queue.Queue):

首先我們繼承 Queue.Queue 類。從 Queue.Queue 類可以繼承 get 和 put 方法，以及隊列的行為。

def__init__(self, num_workers=1): Queue.Queue.__init__(self) self.num_workers=num_workers self.start_workers()

初始化的時候，我們可以不用考慮工作線程的數量。

defadd_task(self, task,*args,**kwargs): args=argsor() kwargs=kwargsor{} self.put((task, args, kwargs))

我們把 task, args, kwargs 以元組的形式存儲在隊列中。*args 可以傳遞數量不等的參數，**kwargs 可以傳遞命名參數。

defstart_workers(self): foriinrange(self.num_workers): t=Thread(target=self.worker) t.daemon=True t.start()

我們為每個 worker 創建一個線程，然後在後台刪除。

下面是 worker 函數的代碼：

defworker(self): whileTrue: tupl=self.get() item, args, kwargs=self.get() item(*args,**kwargs) self.task_done()

worker 函數獲取隊列頂端的任務，並根據輸入參數運行，除此之外，沒有其他的功能。下面是隊列的代碼：

我們可以通過下面的代碼測試：

defblokkah(*args,**kwargs): time.sleep(5) print「Blokkah mofo!」 q=TaskQueue(num_workers=5) foriteminrange(1): q.add_task(blokkah) q.join()# wait for all the tasks to finish. print「Alldone!」

Blokkah 是我們要做的任務名稱。隊列已經緩存在內存中，並且沒有執行很多任務。下面的步驟是把主隊列當做單獨的進程來運行，這樣主程序退出以及執行資料庫持久化時，隊列任務不會停止運行。但是這個例子很好地展示了如何從一個很簡單的小任務寫成像工作隊列這樣複雜的程序。

defgradient_descent(): # the gradient descent code queue.add_task(plotly.write, x=X, y=Y)

修改之後，我的梯度下降演算法工作效率似乎更高了。如果你很感興趣的話，可以參考下面的代碼。fromthreadingimportThreadimportQueueimporttime classTaskQueue(Queue.Queue): def__init__(self, num_workers=1):Queue.Queue.__init__(self)self.num_workers=num_workersself.start_workers() defadd_task(self, task,*args,**kwargs):args=argsor()kwargs=kwargsor{}self.put((task, args, kwargs)) defstart_workers(self):foriinrange(self.num_workers):t=Thread(target=self.worker)t.daemon=Truet.start() defworker(self):whileTrue:tupl=self.get()item, args, kwargs=self.get()item(*args,**kwargs)self.task_done() deftests():defblokkah(*args,**kwargs):time.sleep(5)print”Blokkah mofo!” q=TaskQueue(num_workers=5) foriteminrange(10):q.add_task(blokkah) q.join()# block until all tasks are doneprint”All done!” if__name__==”__main__”:tests()

Python 非同步任務隊列Celery 使用

在 Python 中定義 Celery 的時候，我們要引入 Broker，中文翻譯過來就是「中間人」的意思。在工頭(生產者)提出任務的時候，把所有的任務放到 Broker 裡面，在 Broker 的另外一頭，一群碼農(消費者)等著取出一個個任務準備著手做。這種模式註定了整個系統會是個開環系統，工頭對於碼農們把任務做的怎樣是不知情的。所以我們要引入 Backend 來保存每次任務的結果。這個 Backend 也是存儲任務的信息用的，只不過這裡存的是那些任務的返回結果。我們可以選擇只讓錯誤執行的任務返回結果到 Backend，這樣我們取回結果，便可以知道有多少任務執行失敗了。

其實現架構如下圖所示：

可以看到，Celery 主要包含以下幾個模塊：

celery可以通過pip自動安裝。

broker 可選擇使用RabbitMQ/redis，backend可選擇使用RabbitMQ/redis/MongoDB。RabbitMQ/redis/mongoDB的安裝請參考對應的官方文檔。

——————————rabbitmq相關———————————————————-

官網安裝方法：

啟動管理插件：sbin/rabbitmq-plugins enable rabbitmq_management 啟動rabbitmq：sbin/rabbitmq-server -detached

rabbitmq已經啟動，可以打開頁面來看看 地址：

用戶名密碼都是guest 。進入可以看到具體頁面。 關於rabbitmq的配置，網上很多 自己去搜以下就ok了。

——————————rabbitmq相關——————————————————–

項目結構如下：

使用前，需要三個方面：celery配置，celery實例，需執行的任務函數，如下：

Celery 的配置比較多，可以在 官方配置文檔： 查詢每個配置項的含義。

當然，要保證上述非同步任務and下述定時任務都能正常執行，就需要先啟動celery worker，啟動命令行如下:

需 啟動beat ，執行定時任務時, Celery會通過celery beat進程來完成。Celery beat會保持運行, 一旦到了某一定時任務需要執行時, Celery beat便將其加入到queue中. 不像worker進程, Celery beat只需要一個即可。而且為了避免有重複的任務被發送出去，所以Celery beat僅能有一個。

命令行啟動：

如果你想將celery worker/beat要放到後台運行，推薦可以扔給supervisor。

supervisor.conf如下：

python實現堆棧與隊列的方法

python實現堆棧與隊列的方法

本文實例講述了python實現堆棧與隊列的方法。分享給大家供大家參考。具體分析如下：

1、python實現堆棧，可先將Stack類寫入文件stack.py，在其它程序文件中使用from stack import Stack，然後就可以使用堆棧了。

stack.py的程序:

代碼如下:class Stack():

def __init__(self,size):

self.size=size;

self.stack=[];

self.top=-1;

def push(self,ele): #入棧之前檢查棧是否已滿

if self.isfull():

raise exception(“out of range”);

else:

self.stack.append(ele);

self.top=self.top+1;

def pop(self): # 出棧之前檢查棧是否為空

if self.isempty():

raise exception(“stack is empty”);

else:

self.top=self.top-1;

return self.stack.pop();

def isfull(self):

return self.top+1==self.size;

def isempty(self):

return self.top==-1;

再寫一個程序文件，stacktest.py,使用棧，內容如下:

代碼如下:#!/usr/bin/python

from stack import Stack

s=Stack(20);

for i in range(3):

s.push(i);

s.pop()

print s.isempty();

2、python 實現隊列:

複製代碼代碼如下:class Queue():

def __init__(self,size):

self.size=size;

self.front=-1;

self.rear=-1;

self.queue=[];

def enqueue(self,ele): #入隊操作

if self.isfull():

raise exception(“queue is full”);

else:

self.queue.append(ele);

self.rear=self.rear+1;

def dequeue(self): #出隊操作

if self.isempty():

raise exception(“queue is empty”);

else:

self.front=self.front+1;

return self.queue[self.front];

def isfull(self):

return self.rear-self.front+1==self.size;

def isempty(self):

return self.front==self.rear;

q=Queue(10);

for i in range(3):

q.enqueue(i);

print q.dequeue();

print q.isempty();

希望本文所述對大家的Python程序設計有所幫助。

python 里有沒有輕量的持久化隊列

python標準庫里的queue是非同步隊列。輕量化的隊列可以用collections模塊里的deque。