前言

在Tensorflow的實際應用中,隊列與線程是必不可少,主要應用于數據的加載等,不同的情況下使用不同的隊列,主線程與其他線程異步進行數據的訓練與讀取,所以隊列與線程的知識也是Tensorflow必須要學會的重要知識
另一方面,Tensorflow作為符號編程框架,在構圖后,加載數據有三種方式,預加載與填充數據都存在,數據量大消耗內存等情況的出現.使用第三種方式文件讀取避免了前兩者的缺點,但是其實現則必須要使用隊列與此線程,因此為了進一步的學習,也需要在這個時候掌握隊列與線程的知識

FIFOQueue : 先入先出的隊列

# 在使用循環神經網絡時,希望讀入的訓練樣本是有序的,就要用到FIFOQueue# 先創建一個先入先出的隊列,初始化隊列插入0.1,0.2,0.3三個數字 q = tf.FIFOQueue(3,tf.float32) init = q.enqueue_many(([0.1,0.2,0.3],)) # 定義出隊,+1,入隊操作 x = q.dequeue() y = x+1 q_inc = q.enqueue(y)with tf.Session() as sess:sess.run(init)# quelen = sess.run(q.size())for i in range(2):sess.run(q_inc) # 執行兩次操作,隊列中的值變為0.3,1.1,1.2for j in range(sess.run(q.size())):print(sess.run(q.dequeue())) # 輸出隊列的值

運行結果

0.3 1.1 1.2

RandomShuffleQueue:隨機隊列

隨機隊列,再出隊列時,是以隨機的順序產生元素的 # 例如,我們在訓練一些圖像樣本時,使用CNN的網絡結構,希望可以無序的讀入訓練樣本 q = tf.RandomShuffleQueue(capacity=10,min_after_dequeue=2,dtypes=tf.float32) # capacity 隊列最大長度,min_after_dequeue 出隊后最小的長度 with tf.Session() as sess :for i in range(0,10): # 10次入隊sess.run(q.enqueue(i))for i in range(0,8): # 8次出隊print(sess.run(q.dequeue()))# 在隊列長度等于最小值時,執行出隊操作,會發生阻斷 # 在隊列長度等于最大值時,執行入隊操作,會發生阻斷# 解除阻斷的一種方法---設置等待時間 # run_options = tf.RunOptions(time_out_in_ms = 100000) # 等待十秒 # try: # sess.run(q.dequeue(),options=run_options) # except tf.errors.DeadlineExceededError: # print('out of range')

運行結果:

1.0 6.0 3.0 0.0 7.0 4.0 8.0 9.0

隊列管理器

q = tf.FIFOQueue(1000, tf.float32) counter = tf.Variable(0.0) # 計數器 increment_op = tf.assign_add(counter, tf.constant(1.0)) # 操作給計數器加一 enquence_op = q.enqueue(counter) # 操作: 讓計數器加入隊列# 創建一個隊列管理器QueueRunner,用這兩個操作相對列q中添加元素,目前我們只使用一個線程 qr = tf.train.QueueRunner(q, enqueue_ops=[increment_op, enquence_op] * 1)# 啟動一個會話,從隊列管理器qr中創建線程# 主線程 with tf.Session() as sess:sess.run(tf.global_variables_initializer())enquence_threads = qr.create_threads(sess, start=True) # 啟用入隊線程# 主線程for i in range(10):print(sess.run(q.dequeue()))# tensorflow.python.framework.errors_impl.CancelledError: Run call was cancelled

運行結果:

187.0 190.0 195.0 201.0 206.0 209.0 214.0 219.0 223.0 230.0報錯:......

線程和協調器

使用協調器來管理線程
這是要注意的的是在隊列線程關閉后,再執行出隊操作,將會報錯”tf.errors.OutOfRange”
第一點會報錯很好理解,隊列關閉自然不能出隊了(如果你學過數據結構那么你應該能夠理解隊列的意義,如果沒能理解,建議重新掌握隊列這一基本概念),
第二點則需要注意一下,報的錯誤是超出范圍,而非隊列不存在,希望當你遇到這個錯誤的時候能夠想到,實際上可能是隊列線程已經被取消了,而非真的超出了范圍.

q = tf.FIFOQueue(1000, tf.float32) counter = tf.Variable(0.0) # 計數器 increment_op = tf.assign_add(counter, tf.constant(1.0)) # 操作給計數器加一 enquence_op = q.enqueue(counter) # 操作: 讓計數器加入隊列# 第一種情況,在關閉其他線程之后(除主線程之外的其它線程),調用出隊操作 print('第一種情況,在關閉其他線程之后(除主線程之外的其它線程),調用出隊操作') # 創建一個隊列管理器QueueRunner,用這兩個操作相對列q中添加元素,目前我們只使用一個線程 qr = tf.train.QueueRunner(q, enqueue_ops=[increment_op, enquence_op] * 1)# 主線程 sess = tf.Session() sess.run(tf.global_variables_initializer())# Coordinator: 協調器, 協調線程間的關系,可以視為一種信號量,用來做同步 coord = tf.train.Coordinator()# 啟動入隊線程,協調器是線程的參數 enqueue_threads = qr.create_threads(sess,coord=coord,start=True)# 主線程for i in range(0,10):print(sess.run(q.dequeue()))coord.request_stop() # 通知其他線程關閉 coord.join(enqueue_threads) # join 操作等待其他線程結束,其他所有線程關閉之后,這一函數才能返回print('第二種情況: 在隊列線程關閉之后,調用出隊操作-->處理tf.errors.OutOfRange錯誤')# q啟動入隊線程 enqueue_threads = qr.create_threads(sess,coord=coord,start=True)# 主線程 coord.request_stop() # 通知其他線程關閉for j in range(0,10):try:print(sess.run(q.dequeue()))except tf.errors.OutOfRangeError:breakcoord.join(enqueue_threads) # join 操作等待其他線程結束,其他所有線程關閉之后,這一函數才能返回

運行結果:

第一種情況,在關閉其他線程之后(除主線程之外的其它線程),調用出隊操作 5.0 7.0 10.0 17.0 23.0 29.0 34.0 38.0 46.0 52.0 第二種情況: 在隊列線程關閉之后,調用出隊操作-->處理tf.errors.OutOfRange錯誤 53.0 53.0 53.0 53.0 53.0 53.0 53.0 53.0 53.0 53.0

更詳細的內容你可以參考github中的源代碼

總結

以上是生活随笔為你收集整理的6.1 Tensorflow笔记(基础篇):队列与线程的全部內容，希望文章能夠幫你解決所遇到的問題。

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