1.5 TensorFlow的隊列

隊列（queue）是一種最為常用的數(shù)據(jù)輸入/輸出方式，其通過先進(jìn)先出的線性數(shù)據(jù)結(jié)構(gòu)，一端只負(fù)責(zé)增加TensorFlow的隊列中的數(shù)據(jù)元素，而數(shù)據(jù)的輸出和刪除在隊列的另一端實現(xiàn)。能夠增加數(shù)據(jù)元素的隊列一端被稱為隊尾，而輸出和刪除數(shù)據(jù)元素的一端被稱為隊首。

TensorFlow應(yīng)用隊列作為數(shù)據(jù)的一種基本輸入/輸出方式，可以將新的數(shù)據(jù)插入到隊列的隊尾，而在隊首將數(shù)據(jù)輸出和刪除。當(dāng)然，在TensorFlow中可以這樣認(rèn)為，隊列在TensorFlow中處于一種有狀態(tài)節(jié)點(diǎn)的地位，隨著其他節(jié)點(diǎn)在圖中狀態(tài)的改變，隊列這個“節(jié)點(diǎn)”的狀態(tài)可以隨之改變。

1.5.1 隊列的創(chuàng)建

表1-1為隊列常用的方法匯總。

一般而言，創(chuàng)建一個隊列首先要選定數(shù)據(jù)出入類型。例如，是使用FIFOQueue函數(shù)設(shè)定數(shù)據(jù)為先入先出，還是使用RandomShuffleQueue這種隨機(jī)元素出列的方式。

函數(shù)的第一個參數(shù)是隊列中數(shù)據(jù)的個數(shù)，第二個參數(shù)是隊列中元素的類型。之后要對隊列中元素進(jìn)行初始化和運(yùn)行操作，需要特別注意的是，TensorFlow中任何操作都是在“會話”中進(jìn)行的，因此其基本操作都要由會話（Session）完成。

enqueue_many函數(shù)將上文中創(chuàng)建的FIFOQueue函數(shù)進(jìn)行了填充，因為q被設(shè)置成包含3個元素的函數(shù)，因此其一次性被填充進(jìn)3個數(shù)據(jù)。但是實際上，此時的數(shù)據(jù)填充并沒有完成，而是進(jìn)行了一個預(yù)備工作，真正的工作要在會話中完成，因此還需要運(yùn)行會話中的run函數(shù)。

例1-1為創(chuàng)建隊列實例演示，在實例中首先設(shè)定一個“先入先出”的隊列，之后被填充數(shù)據(jù)。dequeue函數(shù)將其中的數(shù)據(jù)彈出，此時為了能夠讓這個隊列操作完成，這步操作被命名為init2，下面的init3同樣是在對話中完成的。之后通過對話操作對這3個步驟進(jìn)行處理。

【例1-1】創(chuàng)建隊列實例演示。

size函數(shù)獲取了當(dāng)前隊列的數(shù)據(jù)個數(shù)，之后通過一個for循環(huán)將隊列中的數(shù)據(jù)彈出。最終打印結(jié)果如下：

由結(jié)果可以看到，第一次init的3個數(shù)值中0.3被dequeue彈出，取而代之的是enqueue函數(shù)進(jìn)去的1這個數(shù)值。

提示：dequeue是一個可以阻塞隊列的函數(shù)，如果其中沒有數(shù)據(jù)被彈出，則會阻塞隊列直到數(shù)據(jù)填充之后被彈出。

從例1-1中可以看出，隊列操作是在主線程的對話中依次完成的。這樣做的好處是不易阻塞隊列，出了bug容易查找等。例如，數(shù)據(jù)執(zhí)行入隊操作后從硬盤上將數(shù)據(jù)輸入到內(nèi)存中供后續(xù)使用，但是這樣的操作會造成數(shù)據(jù)讀取和輸入較慢，處理相對困難。

TensorFlow中提供了QueueRunner函數(shù)用以解決異步操作問題，如例1-2所示。其可創(chuàng)建一系列的線程同時進(jìn)入主線程內(nèi)進(jìn)行操作，數(shù)據(jù)讀取與操作是同步的，即主線程在進(jìn)行訓(xùn)練模型工作的同時將數(shù)據(jù)從硬盤讀入。

【例1-2】利用QueueRunner函數(shù)解決異步操作問題。

在程序中首先創(chuàng)建了1個數(shù)據(jù)處理函數(shù)，add_op的操作是將整數(shù)1疊加到變量counter上。為了執(zhí)行這個操作，qr創(chuàng)建了一個隊列管理器QueueRunner，其調(diào)用了2個線程去完成此項任務(wù)。create_threads函數(shù)對線程進(jìn)行了啟動，此時線程已經(jīng)開始運(yùn)行。

而在for循環(huán)中，主程序同時也對隊列進(jìn)行操作，即不停地將數(shù)據(jù)從隊列中彈出，結(jié)果如下：

從例1-2中可以看出，程序首先是正確輸出的，但是在后半部分程序執(zhí)行時會報錯。

如果換一種表述形式，結(jié)果會怎樣呢?代碼如下：

可以看到此時的會話并沒有報錯，但是程序也沒有結(jié)束，而是被掛起。造成這種情況的原因是add操作和入隊操作沒有同步，即TensorFlow在隊列設(shè)計時為了優(yōu)化I/O系統(tǒng)，隊列的操作一般使用批處理，這樣入隊線程沒有發(fā)送結(jié)束的信息，而程序主線程期望將程序結(jié)束，因此造成線程阻塞，程序被掛起。

提示：TensorFlow中一般遇到程序被掛起的情況指的是數(shù)據(jù)輸入與處理沒有同步，即需要數(shù)據(jù)時卻沒有數(shù)據(jù)被輸入到隊列中，那么線程就會被整體掛起。而此時tf也不會報錯而是一直處于等待狀態(tài)。

1.5.2 線程同步與停止

可以看到，TensorFlow的會話是支持多線程的，多線程可以很方便地在一個會話下共同工作，并行地相互執(zhí)行。但是通過程序演示也看到，這種同步會造成某個線程想要關(guān)閉對話時，對話被強(qiáng)行關(guān)閉而未完成工作的線程也被強(qiáng)行關(guān)閉。

TensorFlow為了解決多線程的同步和處理問題，提供了Coordinator和QueueRunner函數(shù)來對線程進(jìn)行控制和協(xié)調(diào)。在使用上，這兩個類必須同時工作，共同協(xié)作來停止會話中所有線程，并向等待所有工作的線程終止程序報告。例1-3為線程同步與停止實例演示。

【例1-3】線程同步與停止實例演示。

在程序中，create_threads函數(shù)被添加了一個新的參數(shù)：線程協(xié)調(diào)器，用于協(xié)調(diào)線程之間的關(guān)系。當(dāng)啟動線程后，線程協(xié)調(diào)器在最后負(fù)責(zé)對所有線程接收和處理，當(dāng)一個線程結(jié)束時，線程協(xié)調(diào)器會對所有的線程發(fā)出通知，協(xié)助其完畢。

1.5.3 隊列中數(shù)據(jù)的讀取

TensorFlow支持很多種樣例輸入的方式。第一種方法最容易，即使用placeholder，但這需要手動傳遞numpy.array類型的數(shù)據(jù)；第二種方法就是使用二進(jìn)制文件和輸入隊列的組合形式。這種方法不僅節(jié)省了代碼量，避免了進(jìn)行data augmentation和讀文件操作，可以處理不同類型的數(shù)據(jù)，而且也不再需要人為地劃分開“預(yù)處理”和“模型計算”。在使用TensorFlow進(jìn)行異步計算時，隊列是一種強(qiáng)大的機(jī)制。

正如TensorFlow中的其他組件一樣，隊列就是TensorFlow圖中的節(jié)點(diǎn)。這是一種有狀態(tài)的節(jié)點(diǎn)，就像變量一樣：其他節(jié)點(diǎn)可以修改它的內(nèi)容。具體來說，其他節(jié)點(diǎn)可以把新元素插入到隊尾，也可以把隊首的元素刪除，如FIFOQueue和RandomShuffleQueue等對象。

在TensorFlow中對Tensor進(jìn)行異步計算非常重要。例如，一個典型的輸入結(jié)構(gòu)是使用一個RandomShuffleQueue來作為模型訓(xùn)練輸入的，多線程準(zhǔn)備訓(xùn)練樣本，并且把這些樣本壓入隊列，一個訓(xùn)練線程執(zhí)行一個訓(xùn)練操作，此操作會從隊列中移除最小批次的樣本（mini-batches），這種結(jié)構(gòu)具有許多優(yōu)點(diǎn)。

TensorFlow的Session對象是可以支持多線程的，因此多線程可以很方便地使用同一個會話（Session）并且并行執(zhí)行操作。然而，在Python程序?qū)崿F(xiàn)這樣的并行運(yùn)算卻并不容易。所有線程都必須能被同步終止，異常必須能被正確捕獲并報告。當(dāng)會話終止的時候，隊列必須能被正確地關(guān)閉。所幸TensorFlow提供了兩個類來幫助多線程的實現(xiàn)：tf.Coordinator和 tf.QueueRunner。從設(shè)計上這兩個類必須被一起使用。Coordinator類可以用來同時停止多個工作線程并且向那個在等待所有工作線程終止的程序報告異常。QueueRunner類用來協(xié)調(diào)多個工作線程同時將多個Tensor壓入同一個隊列中。

圖1-6所示為隊列讀取數(shù)據(jù)流程。

在圖1-6中，首先由一個單線程把文件名壓入隊列，兩個Reader同時從隊列中取文件名并讀取數(shù)據(jù)，然后Decoder將讀出的數(shù)據(jù)解碼后壓入樣本隊列，最后單個或批量取出樣本（圖1-6中沒有展示樣本出列）。

此處，搭建數(shù)據(jù)讀取圖需要5個步驟：

（1）將我們的數(shù)據(jù)轉(zhuǎn)成相應(yīng)的文件放入磁盤。

（2）把文件名壓入隊列。

（3）從隊列中取文件名并讀取數(shù)據(jù)。

（4）Decoder將讀出的數(shù)據(jù)解碼。

（5）解碼后壓入樣本隊列，最后單個或批量取出樣本。

下面，就使用代碼一步步來解釋上述過程。

（1）構(gòu)造訓(xùn)練數(shù)據(jù)，使用tf.python_io.TFRecordWriter創(chuàng)建一個專門存儲TensorFlow數(shù)據(jù)的writer，擴(kuò)展名為'.tfrecord'，最后存儲的文件是序列化的文件：

（2）把文件名壓入隊列，從隊列中取文件名并讀取數(shù)據(jù)，Decoder將讀出的數(shù)據(jù)解碼都放在一個函數(shù)里面，與剛才的writer不同，這個reader是符號化的，只有在Session中運(yùn)行才會執(zhí)行：

（3）解碼后壓入樣本隊列，最后單個或批量取出樣本，是使用tf.train.shuffle_batch來實現(xiàn)取出樣本的：