Tensorflow多線程輸入數據處理框架

Tensorflow提供了一系列的對圖像進行預處理的方法，可是複雜的預處理過程會減慢整個訓練過程，因此，爲了不圖像的預處理成爲訓練神經網絡效率的瓶頸，Tensorflow提供了多線程處理輸入數據的框架！

下面將一一闡述多線程處理輸入數據的理論及Tensorflow的源代碼！

 

1 隊列和多線程

首先來介紹Tensorflow中定義隊列的方式，在Tensorflow中，能夠定義FIFOQueue和RandomShuffleQueue兩種類型的隊列，FIFO顧名思義是先入先出型隊列，而RandomShuffleQueue就是隨機讀出的操做，RandomShufleQueue操做在訓練過程當中的做用很大，能夠隨機的從數據集中取出數據，提升魯棒性

 

tensorflow中，隊列和變量相似，都是計算圖上有狀態的節點，其餘的狀態節點能夠修改他們的狀態，對於一個變量來講，賦值操做能夠修改變量的取值，而對於隊列來講，修改隊列的相應狀態主要有三個方法：

1Enquene  Enqueue的操做至關於進入隊列的操做，也就是想要將數據插入到隊列中時，調用此方法，假如已經定義好隊列q，那麼調用q.enqueue(([55]))就能夠將數據寫入隊列

2 EnqueneMany  q.EnqueneMany方法是一次性寫入多個數據！舉例：q.EnqueneMany(([34,55]))

3 Dequeue 此方法至關於從隊列中調出數據的操做， x = q.dequeue()

 

隊列的做用不只僅是一種數據結構，它能夠提供一種多線程的機制，提升數據的讀取存儲速度，好比多線程能夠同時向一個隊列中寫元素，或者同時從一個隊列中讀取元素，多線程並行處理數據，提升數據預處理的速度就是Tensorflow提供的解決上述瓶頸問題的方案！

 

既然是多線程處理，那麼必須須要某個交通指揮官來協調處理多線程的工做，沒錯，協調-coordinator，tensorflow提供了一個tf.Coordinator類和tf.QueueRunner類來完成多線程工做狀態的管理和啓動！

Coordinator提供了三個函數：1 should_stop 2 request_stop 3 join 來進行線程的管理操做，在每一個線程啓動以前，都須要先定義一個coordinator類而後當作參數傳入到線程裏面去，代表接管本線程的全部操做，線程在啓動運行的過程當中不斷的查詢should_stop的狀態，一旦發現should_stop變成True，那麼該線程就必須得退出，一樣，每個啓動的線程均可以調用request_stop函數來通知其餘線程退出，下面給出一段使用這兩個類來操做線程的代碼：

 

 1 import tensorflow as tf 
 2 import numpy as np 
 3 import threading 
 4 import time 
 5 
 6 
 7 
 8 def MyLooP(coord,worker_id):
 9 
10     while not coord.should_stop():
11 
12         if np.random.rand() < 0.1 
13         print("Stoping form id: %d\n" %worker_id)
14         coord.request_stop()
15 
16     else :
17         print("Working on id : %d\n" %worker_id)
18         time.sleep(1)
19 
20 
21 coord = tf.train.coordinator()
22 
23 threads = [ threading.Thread(target = MyLooP,args=(coord,i,)) for i in xrange(5)]
24 
25 for t in threads: t.start()
26 
27 coord.join(threads) #等待全部線程退出
28 
29 #運行以上程序，獲得結果以下：
30 #working on id : 0
31 #working on id : 1
32 #working on id : 2
33 #working on id : 4
34 #working on id : 3
35 #working on id : 0
36 #stoping on id : 4
37 #working on id : 1

 

如上面程序描述的，首先定義了一個MyLoop函數，接受coord和workerid兩個參數，當coord.should_stop != True的時候，產生一個隨機數，若是隨機數小於0.1，那麼對應的這個進程就會去申請request_stop()函數，通知其餘的全部進程結束！

不然的話就打印當前進程的workerid，接下來定義了協調器類的實例coordinator()和五個進程，進程id分別是0-4，以後啓動全部的進程等待某個進程的request_stop()，最後調用一個coord.join()來等待全部進程合理的退出，從運行的結果中咱們

咱們能夠看出，working on id :0,1,2,4,3五個進程分別啓動完成，而後從新進入MyLoop中繼續執行，而後working on id:0, 忽然進程4在執行的過程當中它隨機生成數小於0.1，這致使進程4會去申請request.stop()，從而致使五個進程查詢should_stop所有變成了True，因而5個進程便所有結束自身進程，能夠看到最後一行仍然有working on id:1打印出來，其實是由於在進程4申請Request_stop的過程當中，進程1已經完成了判斷進入到了下一條else中，因此它仍然能夠打印出來，可是再一次循環以前查詢should_stop的時候就執行結束進程的相關操做了！

未完待續.......