參考書

《TensorFlow：實戰Google深度學習框架》（第2版）

對于隊列，修改隊列狀態的操作主要有Enqueue、EnqueueMany和Dequeue。以下程序展示了如何使用這些函數來操作一個隊列。

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# coding=utf-8 """
@author: Li Tian
@contact: 694317828@qq.com
@software: pycharm
@file: queue_operate.py
@time: 2019/1/31 21:32
@desc: 操作一個隊列
"""import tensorflow as tf# 創建一個先進先出的隊列，指定隊列中最多可以保存兩個元素，并指定類型為整數
q = tf.FIFOQueue(2, "int32")
# 使用enqueue_many函數來初始化隊列中的元素。和變量初始化類似，在使用隊列之前需要明確的調用這個初始化過程。
init = q.enqueue_many(([0, 10],))
# 使用Dequeue函數將隊列中的第一個元素出隊列。這個元素的值將被存在變量x中
x = q.dequeue()
# 將得到的值+1
y = x + 1
# 將+1后的值再重新加入隊列。
q_inc = q.enqueue([y])with tf.Session() as sess:# 運行初始化隊列的操作
    init.run()for _ in range(5):# 運行q_inc將執行數據出隊列、出隊的元素+1、重新加入隊列的整個過程。v, _ = sess.run([x, q_inc])# 打印出隊元素的取值print(v)

運行結果：

?

tf.Coordinator主要用于協同多個線程一起停止，以下程序展示了如何使用tf.Coordinator。

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# coding=utf-8 """
@author: Li Tian
@contact: 694317828@qq.com
@software: pycharm
@file: coordinator_test1.py
@time: 2019/2/2 21:35
@desc: tf.Coordinator主要用于協同多個線程一起停止，以下程序展示了如何使用tf.Coordinator
"""import tensorflow as tf
import numpy as np
import threading
import time# 線程中運行的程序，這個程序每隔1秒判斷是否需要停止并打印自己的ID。
def MyLoop(coord, worker_id):# 使用tf.Coordinator類提供的協同工具判斷當前線程是否需要停止while not coord.should_stop():# 隨機停止所有的線程。if np.random.rand() < 0.1:print("Stoping from id: %d\n" % worker_id)# 調用coord.request_stop()函數來通知其他線程停止。
            coord.request_stop()else:# 打印當前線程的Id。print("Working on id: %d\n" % worker_id)# 暫停1秒time.sleep(1)# 聲明一個tf.train.Coordinator類來協同多個線程。
coord = tf.train.Coordinator()
# 聲明創建5個線程。
threads = [threading.Thread(target=MyLoop, args=(coord, i, )) for i in range(5)]
# 啟動所有的線程
for t in threads:t.start()
# 等待所有線程退出
coord.join(threads)

?運行結果：

?

?如何使用tf.QueueRunner和tf.Coordinator來管理多線程隊列操作。

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# coding=utf-8 """
@author: Li Tian
@contact: 694317828@qq.com
@software: pycharm
@file: queuerunner_test1.py
@time: 2019/2/3 12:31
@desc: 如何使用tf.QueueRunner和tf.Coordinator來管理多線程隊列操作。
"""import tensorflow as tf# 聲明一個先進先出的隊列，隊列中最多100個元素，類型為實數
queue = tf.FIFOQueue(100, "float")
# 定義隊列的入隊操作
enqueue_op = queue.enqueue([tf.random_normal([1])])# 使用tf.train.QueueRunner來創建多個線程運行隊列的入隊操作。
# tf.train.QueueRunner的第一個參數給出了被操作的隊列，[enqueue_op] * 5
# 表示了需要啟動5個線程，每個線程中運行的是enqueue_op操作
qr = tf.train.QueueRunner(queue, [enqueue_op] * 5)# 將定義過的QueueRunner加入Tensorflow計算圖上指定的集合。
# tf.train.add_queue_runner函數沒有指定集合
# 則加入默認集合tf.GraphKeys.QUEUE_RUNNERS。下面的函數就是將剛剛定義的
# qr加入默認的tf.GraphKeys.QUEUE_RUNNER集合。
tf.train.add_queue_runner(qr)
# 定義出隊操作
out_tensor = queue.dequeue()with tf.Session() as sess:# 使用tf.train.Coordinator來協同啟動的線程。coord = tf.train.Coordinator()# 使用tf.train.QueueRunner時，需要明確調用tf.train.start_queue_runners# 來啟動所有線程。否則因為沒有線程運行入隊操作，當調用出隊操作的時候，程序會一直# 等待入隊操作被運行。tf.train.start_queue_runners函數會默認啟動# tf.GraphKeys.QUEUE_RUNNERS集合中所有的QueueRunner。因為這個函數值支持啟動# 指定集合中的QueueRunner，所以一般來說tf.train.add_queue_runner函數和# tf.trian.start_queue_runners函數會指定同一個集合。threads = tf.train.start_queue_runners(sess=sess, coord=coord)# 獲取隊列中的取值。for _ in range(3):print(sess.run(out_tensor)[0])# 使用tf.train.Coordinator來停止所有的線程
    coord.request_stop()coord.join(threads)

??運行結果：

?

轉載于:https://www.cnblogs.com/lyjun/p/10350282.html

總結

以上是生活随笔為你收集整理的Tensorflow多线程输入数据处理框架（一）——队列与多线程的全部內容，希望文章能夠幫你解決所遇到的問題。

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