java中的本義字符引見

比來我在用梯度降低算法繪制神經收集的數據時，碰到了一些算法機能的成績。梯度降低算法的代碼以下（偽代碼）：

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

普通來講，當收集要求 plot.ly 畫圖時會壅塞期待前往，因而也會影響到其他的梯度降低函數的履行速度。

一種處理方法是每挪用一次 plotly.write 函數就開啟一個新的線程，然則這類辦法感到不是很好。 我不想用一個像 cerely（一種散布式義務隊列）一樣年夜而全的義務隊列框架，由於框架關於我的這點需求來講太重了，而且我的畫圖也其實不須要 redis 來耐久化數據。

那用甚麼方法處理呢？我在 python 中寫了一個很小的義務隊列，它可以在一個零丁的線程中挪用 plotly.write函數。上面是法式代碼。

from threading import Thread
import Queue 
import time

class TaskQueue(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()

初始化的時刻，我們可以不消斟酌任務線程的數目。

def add_task(self, task, *args, **kwargs):
  args = args or ()
  kwargs = kwargs or {}
  self.put((task, args, kwargs))

我們把 task, args, kwargs 以元組的情勢存儲在隊列中。*args 可以傳遞數目不等的參數，**kwargs 可以傳遞定名參數。

def start_workers(self):
  for i in range(self.num_workers):
    t = Thread(target=self.worker)
    t.daemon = True
    t.start()

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

上面是 worker 函數的代碼：

def worker(self):
  while True:
    tupl = self.get()
    item, args, kwargs = self.get()
    item(*args, **kwargs) 
    self.task_done()

worker 函數獲得隊列頂真個義務，並依據輸出參數運轉，除此以外，沒有其他的功效。上面是隊列的代碼：

我們可以經由過程上面的代碼測試：

def blokkah(*args, **kwargs):
  time.sleep(5)
  print “Blokkah mofo!”

q = TaskQueue(num_workers=5)

for item in range(1):
  q.add_task(blokkah)

q.join() # wait for all the tasks to finish.

print “All done!”

Blokkah 是我們要做的義務稱號。隊列曾經緩存在內存中，而且沒有履行許多義務。上面的步調是把客隊列當作零丁的過程來運轉，如許主法式加入和履行數據庫耐久化時，隊列義務不會停滯運轉。然則這個例子很好地展現了若何從一個很簡略的小義務寫成像任務隊列如許龐雜的法式。

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

修正以後，我的梯度降低算法任務效力仿佛更高了。假如你很感興致的話，可以參考上面的代碼。

from threading import Thread
import Queue
import time

class TaskQueue(Queue.Queue):

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

def add_task(self, task, *args, **kwargs):
args = args or ()
kwargs = kwargs or {}
self.put((task, args, kwargs))

def start_workers(self):
for i in range(self.num_workers):
t = Thread(target=self.worker)
t.daemon = True
t.start()

def worker(self):
while True:
tupl = self.get()
item, args, kwargs = self.get()
item(*args, **kwargs)
self.task_done()

def tests():
def blokkah(*args, **kwargs):
time.sleep(5)
print "Blokkah mofo!"

q = TaskQueue(num_workers=5)

for item in range(10):
q.add_task(blokkah)

q.join() # block until all tasks are done
print "All done!"

if __name__ == "__main__":
tests()