前言

在Tensorflow的實(shí)際應(yīng)用中,隊(duì)列與線程是必不可少,主要應(yīng)用于數(shù)據(jù)的加載等,不同的情況下使用不同的隊(duì)列,主線程與其他線程異步進(jìn)行數(shù)據(jù)的訓(xùn)練與讀取,所以隊(duì)列與線程的知識(shí)也是Tensorflow必須要學(xué)會(huì)的重要知識(shí)
另一方面,Tensorflow作為符號(hào)編程框架,在構(gòu)圖后,加載數(shù)據(jù)有三種方式,預(yù)加載與填充數(shù)據(jù)都存在,數(shù)據(jù)量大消耗內(nèi)存等情況的出現(xiàn).使用第三種方式文件讀取避免了前兩者的缺點(diǎn),但是其實(shí)現(xiàn)則必須要使用隊(duì)列與此線程,因此為了進(jìn)一步的學(xué)習(xí),也需要在這個(gè)時(shí)候掌握隊(duì)列與線程的知識(shí)

FIFOQueue : 先入先出的隊(duì)列

# 在使用循環(huán)神經(jīng)網(wǎng)絡(luò)時(shí),希望讀入的訓(xùn)練樣本是有序的,就要用到FIFOQueue# 先創(chuàng)建一個(gè)先入先出的隊(duì)列,初始化隊(duì)列插入0.1,0.2,0.3三個(gè)數(shù)字 q = tf.FIFOQueue(3,tf.float32) init = q.enqueue_many(([0.1,0.2,0.3],)) # 定義出隊(duì),+1,入隊(duì)操作 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) # 執(zhí)行兩次操作,隊(duì)列中的值變?yōu)?.3,1.1,1.2for j in range(sess.run(q.size())):print(sess.run(q.dequeue())) # 輸出隊(duì)列的值

運(yùn)行結(jié)果

0.3 1.1 1.2

RandomShuffleQueue:隨機(jī)隊(duì)列

隨機(jī)隊(duì)列,再出隊(duì)列時(shí),是以隨機(jī)的順序產(chǎn)生元素的 # 例如,我們?cè)谟?xùn)練一些圖像樣本時(shí),使用CNN的網(wǎng)絡(luò)結(jié)構(gòu),希望可以無(wú)序的讀入訓(xùn)練樣本 q = tf.RandomShuffleQueue(capacity=10,min_after_dequeue=2,dtypes=tf.float32) # capacity 隊(duì)列最大長(zhǎng)度,min_after_dequeue 出隊(duì)后最小的長(zhǎng)度 with tf.Session() as sess :for i in range(0,10): # 10次入隊(duì)sess.run(q.enqueue(i))for i in range(0,8): # 8次出隊(duì)print(sess.run(q.dequeue()))# 在隊(duì)列長(zhǎng)度等于最小值時(shí),執(zhí)行出隊(duì)操作,會(huì)發(fā)生阻斷 # 在隊(duì)列長(zhǎng)度等于最大值時(shí),執(zhí)行入隊(duì)操作,會(huì)發(fā)生阻斷# 解除阻斷的一種方法---設(shè)置等待時(shí)間 # 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')

運(yùn)行結(jié)果:

1.0 6.0 3.0 0.0 7.0 4.0 8.0 9.0

隊(duì)列管理器

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

運(yùn)行結(jié)果:

187.0 190.0 195.0 201.0 206.0 209.0 214.0 219.0 223.0 230.0報(bào)錯(cuò):......

線程和協(xié)調(diào)器

使用協(xié)調(diào)器來(lái)管理線程
這是要注意的的是在隊(duì)列線程關(guān)閉后,再執(zhí)行出隊(duì)操作,將會(huì)報(bào)錯(cuò)”tf.errors.OutOfRange”
第一點(diǎn)會(huì)報(bào)錯(cuò)很好理解,隊(duì)列關(guān)閉自然不能出隊(duì)了(如果你學(xué)過(guò)數(shù)據(jù)結(jié)構(gòu)那么你應(yīng)該能夠理解隊(duì)列的意義,如果沒(méi)能理解,建議重新掌握隊(duì)列這一基本概念),
第二點(diǎn)則需要注意一下,報(bào)的錯(cuò)誤是超出范圍,而非隊(duì)列不存在,希望當(dāng)你遇到這個(gè)錯(cuò)誤的時(shí)候能夠想到,實(shí)際上可能是隊(duì)列線程已經(jīng)被取消了,而非真的超出了范圍.

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

運(yùn)行結(jié)果:

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

更詳細(xì)的內(nèi)容你可以參考github中的源代碼

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。