熱線電話：13121318867

登錄
首頁精彩閱讀Python實現簡單多線程任務隊列
Python實現簡單多線程任務隊列
2018-02-19
收藏

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

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


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

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

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

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

fromthreadingimportThread
importQueue
importtime
 
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)

修改之后，我的梯度下降算法工作效率似乎更高了。如果你很感興趣的話，可以參考下面的代碼。

fromthreadingimportThread
importQueue
importtime
 
classTaskQueue(Queue.Queue):
 
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))
 
defstart_workers(self):
foriinrange(self.num_workers):
t=Thread(target=self.worker)
t.daemon=True
t.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 done
print"All done!"
 
if__name__=="__main__":
tests()




CDA數據分析師考試相關入口一覽（建議收藏）：

? 想報名CDA認證考試，點擊>>>   “CDA報名” 了解CDA考試詳情；

? 想學習CDA考試教材，點擊>>> “CDA教材” 了解CDA考試詳情；

? 想加入CDA考試題庫，點擊>>> “CDA題庫” 了解CDA考試詳情；

? 想了解CDA考試含金量，點擊>>> “CDA含金量” 了解CDA考試詳情；

梯度下降 神經網絡 python

數據分析咨詢請掃描二維碼

若不方便掃碼，搜微信號：CDAshujufenxi

上一篇回歸系列（一）| 怎樣正確地理解線性回歸
下一篇2020美國總統競選大戲開鑼，川普當選的奇跡會再發生嗎？
數據分析師考試動態
CDA報考指南
數據分析學習
數據分析師資訊
更多

CDA教育閉環

常見問題

關于我們

CDA數據分析師公眾號

  • CDA數據分析師公眾號

CDA考試中心小程序

  • CDA考試中心服務號

CDA數據分析師App下載

  • CDA數據分析師App下載
Copyright © 2015-2021, www.ruiqisteel.com All Rights Reserved. CDA數據分析師(北京國富如荷網絡科技有限公司) 版權所有 京ICP備11001960號-9 京公網安備 11010802034615號 經營許可證編號：京B2-20210330
聯系電話：13321103290 (微信同號)

OK
客服在線
立即咨詢
日韩人妻系列无码专区视频,先锋高清无码,无码免费视欧非,国精产品一区一区三区无码
客服在線
立即咨詢