一、線程和隊列

在使用TensorFlow進行異步計算時，隊列是一種強大的機制。

為了感受一下隊列，讓我們來看一個簡單的例子。我們先創建一個“先入先出”的隊列（FIFOQueue），并將其內部所有元素初始化為零。然后，我們構建一個TensorFlow圖，它從隊列前端取走一個元素，加上1之后，放回隊列的后端。慢慢地，隊列的元素的值就會增加。

TensorFlow提供了兩個類來幫助多線程的實現：tf.Coordinator和 tf.QueueRunner。Coordinator類可以用來同時停止多個工作線程并且向那個在等待所有工作線程終止的程序報告異常，QueueRunner類用來協調多個工作線程同時將多個張量推入同一個隊列中。

隊列概述

隊列，如FIFOQueue和RandomShuffleQueue，在TensorFlow的張量異步計算時都非常重要。

例如，一個典型的輸入結構：是使用一個RandomShuffleQueue來作為模型訓練的輸入：

• 多個線程準備訓練樣本，并且把這些樣本推入隊列。
• 一個訓練線程執行一個訓練操作

同步執行隊列

# 創建一個隊列 Q = tf.FIFOQueue(3, dtypes=tf.float32)# 數據進隊列 init = Q.enqueue_many(([0.1, 0.2, 0.3],))# 定義操作,op，出隊列，+1，進隊列,注意返回的都是op out_q = Q.dequeue() data = out_q + 1 en_q = Q.enqueue(data)with tf.Session() as sess:# 初始化隊列，是數據進入sess.run(init)# 執行兩次入隊加1for i in range(2):sess.run(en_q)# 循環取隊列for i in range(3):print(sess.run(Q.dequeue()))

tf.QueueRunner

QueueRunner類會創建一組線程， 這些線程可以重復的執行Enquene操作， 他們使用同一個Coordinator來處理線程同步終止。此外，一個QueueRunner會運行一個closer thread，當Coordinator收到異常報告時，這個closer thread會自動關閉隊列。

您可以使用一個queue runner，來實現上述結構。 首先建立一個TensorFlow圖表，這個圖表使用隊列來輸入樣本。增加處理樣本并將樣本推入隊列中的操作。增加training操作來移除隊列中的樣本。

tf.Coordinator

Coordinator類用來幫助多個線程協同工作，多個線程同步終止。 其主要方法有：

• should_stop():如果線程應該停止則返回True。
• request_stop(): 請求該線程停止。
• join():等待被指定的線程終止。

首先創建一個Coordinator對象，然后建立一些使用Coordinator對象的線程。這些線程通常一直循環運行，一直到should_stop()返回True時停止。 任何線程都可以決定計算什么時候應該停止。它只需要調用request_stop()，同時其他線程的should_stop()將會返回True，然后都停下來。

異步執行隊列：

#主線程，不斷的去取數據，開啟其它線程來進行增加計數，入隊 #主線程結束了，隊列線程沒有結束，就會拋出異常 #主線程沒有結束，需要將隊列線程關閉，防止主線程等待Q = tf.FIFOQueue(1000,dtypes=tf.float32)# 定義操作 var = tf.Variable(0.0) increment_op = tf.assign_add(var,tf.constant(1.0)) en_op = Q.enqueue(increment_op)# 創建一個隊列管理器，指定線程數，執行隊列的操作 qr = tf.train.QueueRunner(Q,enqueue_ops=[increment_op,en_op]*3)with tf.Session() as sess:tf.global_variables_initializer().run()# 生成一個線程協調器coord = tf.train.Coordinator()# 啟動線程執行操作threads_list = qr.create_threads(sess,coord=coord,start=True)print(len(threads_list),"----------")# 主線程去取數據for i in range(20):print(sess.run(Q.dequeue()))# 請求其它線程終止coord.request_stop()# 關閉線程coord.join(threads_list)

二、讀取數據

小數量數據讀取

這僅用于可以完全加載到存儲器中的小的數據集有兩種方法：

• 存儲在常數中。
• 存儲在變量中，初始化后，永遠不要改變它的值。

使用常數更簡單一些，但是會使用更多的內存，因為常數會內聯的存儲在數據流圖數據結構中，這個結構體可能會被復制幾次。

training_data = ... training_labels = ... with tf.Session():input_data = tf.constant(training_data)input_labels = tf.constant(training_labels)

要改為使用變量的方式，您就需要在數據流圖建立后初始化這個變量。

training_data = ... training_labels = ... with tf.Session() as sess:data_initializer = tf.placeholder(dtype=training_data.dtype,shape=training_data.shape)label_initializer = tf.placeholder(dtype=training_labels.dtype,shape=training_labels.shape)input_data = tf.Variable(data_initalizer, trainable=False, collections=[])input_labels = tf.Variable(label_initalizer, trainable=False, collections=[])...sess.run(input_data.initializer,feed_dict={data_initializer: training_data})sess.run(input_labels.initializer,feed_dict={label_initializer: training_lables})

設定trainable=False可以防止該變量被數據流圖的GraphKeys.TRAINABLE_VARIABLES收集，這樣我們就不會在訓練的時候嘗試更新它的值;設定collections=[]可以防止GraphKeys.VARIABLES收集后做為保存和恢復的中斷點。設定這些標志，是為了減少額外的開銷

文件讀取

先看下文件讀取以及讀取數據處理成張量結果的過程：

一般數據文件格式有文本、excel和圖片數據。那么TensorFlow都有對應的解析函數，除了這幾種。還有TensorFlow指定的文件格式。

標準TensorFlow格式

TensorFlow還提供了一種內置文件格式TFRecord，二進制數據和訓練類別標簽數據存儲在同一文件。模型訓練前圖像等文本信息轉換為TFRecord格式。TFRecord文件是protobuf格式。數據不壓縮，可快速加載到內存。TFRecords文件包含 tf.train.Example protobuf，需要將Example填充到協議緩沖區，將協議緩沖區序列化為字符串，然后使用該文件將該字符串寫入TFRecords文件。在圖像操作我們會介紹整個過程以及詳細參數。

數據讀取實現

文件隊列生成函數

• tf.train.string_input_producer(string_tensor, num_epochs=None, shuffle=True, seed=None, capacity=32, name=None)

產生指定文件張量

文件閱讀器類

• class tf.TextLineReader

閱讀文本文件逗號分隔值（CSV）格式

• tf.FixedLengthRecordReader

要讀取每個記錄是固定數量字節的二進制文件

• tf.TFRecordReader

讀取TfRecords文件

解碼

由于從文件中讀取的是字符串，需要函數去解析這些字符串到張量

• tf.decode_csv（records，record_defaults,field_delim = None，name = None）將CSV轉換為張量，與tf.TextLineReader搭配使用

• tf.decode_raw（bytes，out_type,little_endian = None，name = None） 將字節轉換為一個數字向量表示，字節為一字符串類型的張量,與函數tf.FixedLengthRecordReader搭配使用

生成文件隊列

將文件名列表交給tf.train.string_input_producer函數。string_input_producer來生成一個先入先出的隊列，文件閱讀器會需要它們來取數據。string_input_producer提供的可配置參數來設置文件名亂序和最大的訓練迭代數，QueueRunner會為每次迭代（epoch）將所有的文件名加入文件名隊列中，如果shuffle=True的話，會對文件名進行亂序處理。一過程是比較均勻的，因此它可以產生均衡的文件名隊列。

這個QueueRunner工作線程是獨立于文件閱讀器的線程，因此亂序和將文件名推入到文件名隊列這些過程不會阻塞文件閱讀器運行。根據你的文件格式，選擇對應的文件閱讀器，然后將文件名隊列提供給閱讀器的 read 方法。閱讀器的read方法會輸出一個鍵來表征輸入的文件和其中紀錄（對于調試非常有用），同時得到一個字符串標量，這個字符串標量可以被一個或多個解析器，或者轉換操作將其解碼為張量并且構造成為樣本。

# 讀取CSV格式文件 # 1、構建文件隊列# 2、構建讀取器，讀取內容# 3、解碼內容# 4、現讀取一個內容，如果有需要，就批處理內容 import tensorflow as tf import os def readcsv_decode(filelist):"""讀取并解析文件內容:param filelist: 文件列表:return: None"""# 把文件目錄和文件名合并flist = [os.path.join("./csvdata/",file) for file in filelist]# 構建文件隊列file_queue = tf.train.string_input_producer(flist,shuffle=False)# 構建閱讀器，讀取文件內容reader = tf.TextLineReader()key,value = reader.read(file_queue)record_defaults = [["null"],["null"]] # [[0],[0],[0],[0]]# 解碼內容，按行解析，返回的是每行的列數據example,label = tf.decode_csv(value,record_defaults=record_defaults)# 通過tf.train.batch來批處理數據example_batch,label_batch = tf.train.batch([example,label],batch_size=9,num_threads=1,capacity=9)with tf.Session() as sess:# 線程協調員coord = tf.train.Coordinator()# 啟動工作線程threads = tf.train.start_queue_runners(sess,coord=coord)# 這種方法不可取# for i in range(9):# print(sess.run([example,label]))# 打印批處理的數據print(sess.run([example_batch,label_batch]))coord.request_stop()coord.join(threads)return Noneif __name__=="__main__":filename_list = os.listdir("./csvdata")readcsv_decode(filename_list)

每次read的執行都會從文件中讀取一行內容，注意，(這與后面的圖片和TfRecords讀取不一樣)，decode_csv操作會解析這一行內容并將其轉為張量列表。如果輸入的參數有缺失，record_default參數可以根據張量的類型來設置默認值。在調用run或者eval去執行read之前，你必須調用tf.train.start_queue_runners來將文件名填充到隊列。否則read操作會被阻塞到文件名隊列中有值為止。

三、圖像操作

圖像基本概念

在圖像數字化表示當中，分為黑白和彩色兩種。在數字化表示圖片的時候，有三個因素。分別是圖片的長、圖片的寬、圖片的顏色通道數。那么黑白圖片的顏色通道數為1，它只需要一個數字就可以表示一個像素位；而彩色照片就不一樣了，它有三個顏色通道，分別為RGB，通過三個數字表示一個像素位。TensorFlow支持JPG、PNG圖像格式，RGB、RGBA顏色空間。圖像用與圖像尺寸相同(heightwidthchnanel)張量表示。圖像所有像素存在磁盤文件，需要被加載到內存。

圖像大小壓縮

大尺寸圖像輸入占用大量系統內存。訓練CNN需要大量時間，加載大文件增加更多訓練時間，也難存放多數系統GPU顯存。大尺寸圖像大量無關本征屬性信息，影響模型泛化能力。最好在預處理階段完成圖像操作，縮小、裁剪、縮放、灰度調整等。圖像加載后，翻轉、扭曲，使輸入網絡訓練信息多樣化，緩解過擬合。Python圖像處理框架PIL、OpenCV。TensorFlow提供部分圖像處理方法。

• tf.image.resize_images 壓縮圖片導致定大小

圖像數據讀取實例

同樣圖像加載與二進制文件相同。圖像需要解碼。輸入生成器(tf.train.string_input_producer)找到所需文件，加載到隊列。tf.WholeFileReader 加載完整圖像文件到內存，WholeFileReader.read 讀取圖像，tf.image.decode_jpeg 解碼JPEG格式圖像。圖像是三階張量。RGB值是一階張量。加載圖像格式為[batch_size,image_height,image_width,channels]。批數據圖像過大過多，占用內存過高，系統會停止響應。直接加載TFRecord文件，可以節省訓練時間。支持寫入多個樣本。

讀取圖片數據到Tensor

管道讀端多文件內容處理

但是會發現read只返回一個圖片的值。所以我們在之前處理文件的整個流程中，后面的內容隊列的出隊列需要用特定函數去獲取。

• tf.train.batch 讀取指定大小（個數）的張量
• tf.train.shuffle_batch 亂序讀取指定大小（個數）的張量

def readpic_decode(file_list):"""批量讀取圖片并轉換成張量格式:param file_list: 文件名目錄列表:return: None"""# 構造文件隊列file_queue = tf.train.string_input_producer(file_list)# 圖片閱讀器和讀取數據reader = tf.WholeFileReader()key,value = reader.read(file_queue)# 解碼成張量形式image_first = tf.image.decode_jpeg(value)print(image_first)# 縮小圖片到指定長寬，不用指定通道數image = tf.image.resize_images(image_first,[256,256])# 設置圖片的靜態形狀image.set_shape([256,256,3])print(image)# 批處理圖片數據，tensors是需要具體的形狀大小image_batch = tf.train.batch([image],batch_size=100,num_threads=1,capacity=100)tf.summary.image("pic",image_batch)with tf.Session() as sess:merged = tf.summary.merge_all()filewriter = tf.summary.FileWriter("/tmp/summary/dog/",graph=sess.graph)# 線程協調器coord = tf.train.Coordinator()# 開啟線程threads = tf.train.start_queue_runners(sess=sess,coord=coord)print(sess.run(image_batch))summary = sess.run(merged)filewriter.add_summary(summary)# 等待線程回收coord.request_stop()coord.join(threads)return Noneif __name__=="__main__":# 獲取文件列表filename = os.listdir("./dog/")# 組合文件目錄和文件名file_list = [os.path.join("./dog/",file) for file in filename]# 調用讀取函數readpic_decode(file_list)

讀取TfRecords文件數據

#CIFAR-10的數據讀取以及轉換成TFRecordsg格式#1、數據的讀取FLAGS = tf.app.flags.FLAGStf.app.flags.DEFINE_string("data_dir","./cifar10/cifar-10-batches-bin/","CIFAR數據目錄") tf.app.flags.DEFINE_integer("batch_size",50000,"樣本個數") tf.app.flags.DEFINE_string("records_file","./cifar10/cifar.tfrecords","tfrecords文件位置")class CifarRead(object):def __init__(self,filename):self.filelist = filename# 定義圖片的長、寬、深度，標簽字節，圖像字節，總字節數self.height = 32self.width = 32self.depth = 3self.label_bytes = 1self.image_bytes = self.height*self.width*self.depthself.bytes = self.label_bytes + self.image_bytesdef readcifar_decode(self):"""讀取數據，進行轉換:return: 批處理的圖片和標簽"""# 1、構造文件隊列file_queue = tf.train.string_input_producer(self.filelist)# 2、構造讀取器，讀取內容reader = tf.FixedLengthRecordReader(self.bytes)key,value = reader.read(file_queue)# 3、文件內容解碼image_label = tf.decode_raw(value,tf.uint8)# 分割標簽與圖像張量，轉換成相應的格式label = tf.cast(tf.slice(image_label,[0],[self.label_bytes]),tf.int32)image = tf.slice(image_label,[self.label_bytes],[self.image_bytes])print(image)# 給image設置形狀，防止批處理出錯image_tensor = tf.reshape(image,[self.height,self.width,self.depth])print(image_tensor.eval())# depth_major = tf.reshape(image, [self.depth,self.height, self.width])# image_tensor = tf.transpose(depth_major, [1, 2, 0])# 4、處理流程image_batch,label_batch = tf.train.batch([image_tensor,label],batch_size=10,num_threads=1,capacity=10)return image_batch,label_batchdef convert_to_tfrecords(self,image_batch,label_batch):"""轉換成TFRecords文件:param image_batch: 圖片數據Tensor:param label_batch: 標簽數據Tensor:param sess: 會話:return: None"""# 創建一個TFRecord存儲器writer = tf.python_io.TFRecordWriter(FLAGS.records_file)# 構造每個樣本的Examplefor i in range(10):print("---------")image = image_batch[i]# 將單個圖片張量轉換為字符串，以可以存進二進制文件image_string = image.eval().tostring()# 使用eval需要注意的是，必須存在會話上下文環境label = int(label_batch[i].eval()[0])# 構造協議塊example = tf.train.Example(features=tf.train.Features(feature={"image": tf.train.Feature(bytes_list=tf.train.BytesList(value=[image_string])),"label": tf.train.Feature(int64_list=tf.train.Int64List(value=[label]))}))# 寫進文件writer.write(example.SerializeToString())writer.close()return Nonedef read_from_tfrecords(self):"""讀取tfrecords:return: None"""file_queue = tf.train.string_input_producer(["./cifar10/cifar.tfrecords"])reader = tf.TFRecordReader()key, value = reader.read(file_queue)features = tf.parse_single_example(value, features={"image":tf.FixedLenFeature([], tf.string),"label":tf.FixedLenFeature([], tf.int64),})image = tf.decode_raw(features["image"], tf.uint8)# 設置靜態形狀，可用于轉換動態形狀image.set_shape([self.image_bytes])print(image)image_tensor = tf.reshape(image,[self.height,self.width,self.depth])print(image_tensor)label = tf.cast(features["label"], tf.int32)print(label)image_batch, label_batch = tf.train.batch([image_tensor, label],batch_size=10,num_threads=1,capacity=10)print(image_batch)print(label_batch)with tf.Session() as sess:coord = tf.train.Coordinator()threads = tf.train.start_queue_runners(sess=sess,coord=coord)print(sess.run([image_batch, label_batch]))coord.request_stop()coord.join(threads)return Noneif __name__=="__main__":# 構造文件名字的列表filename = os.listdir(FLAGS.data_dir)file_list = [os.path.join(FLAGS.data_dir, file) for file in filename if file[-3:] == "bin"]cfar = CifarRead(file_list)# image_batch,label_batch = cfar.readcifar_decode()cfar.read_from_tfrecords()with tf.Session() as sess:# 構建線程協調器coord = tf.train.Coordinator()# 開啟線程threads = tf.train.start_queue_runners(sess=sess, coord=coord)# print(sess.run(image_batch))# 存進文件# cfar.convert_to_tfrecords(image_batch, label_batch)coord.request_stop()coord.join(threads)

總結

以上是生活随笔為你收集整理的机器学习Tensorflow基本操作：线程队列图像的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。