參考書

《TensorFlow：實(shí)戰(zhàn)Google深度學(xué)習(xí)框架》（第2版）

對(duì)于隊(duì)列，修改隊(duì)列狀態(tài)的操作主要有Enqueue、EnqueueMany和Dequeue。以下程序展示了如何使用這些函數(shù)來(lái)操作一個(gè)隊(duì)列。

#!/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: 操作一個(gè)隊(duì)列
"""import tensorflow as tf# 創(chuàng)建一個(gè)先進(jìn)先出的隊(duì)列，指定隊(duì)列中最多可以保存兩個(gè)元素，并指定類型為整數(shù)
q = tf.FIFOQueue(2, "int32")
# 使用enqueue_many函數(shù)來(lái)初始化隊(duì)列中的元素。和變量初始化類似，在使用隊(duì)列之前需要明確的調(diào)用這個(gè)初始化過(guò)程。
init = q.enqueue_many(([0, 10],))
# 使用Dequeue函數(shù)將隊(duì)列中的第一個(gè)元素出隊(duì)列。這個(gè)元素的值將被存在變量x中
x = q.dequeue()
# 將得到的值+1
y = x + 1
# 將+1后的值再重新加入隊(duì)列。
q_inc = q.enqueue([y])with tf.Session() as sess:# 運(yùn)行初始化隊(duì)列的操作
    init.run()for _ in range(5):# 運(yùn)行q_inc將執(zhí)行數(shù)據(jù)出隊(duì)列、出隊(duì)的元素+1、重新加入隊(duì)列的整個(gè)過(guò)程。v, _ = sess.run([x, q_inc])# 打印出隊(duì)元素的取值print(v)

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

?

tf.Coordinator主要用于協(xié)同多個(gè)線程一起停止，以下程序展示了如何使用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主要用于協(xié)同多個(gè)線程一起停止，以下程序展示了如何使用tf.Coordinator
"""import tensorflow as tf
import numpy as np
import threading
import time# 線程中運(yùn)行的程序，這個(gè)程序每隔1秒判斷是否需要停止并打印自己的ID。
def MyLoop(coord, worker_id):# 使用tf.Coordinator類提供的協(xié)同工具判斷當(dāng)前線程是否需要停止while not coord.should_stop():# 隨機(jī)停止所有的線程。if np.random.rand() < 0.1:print("Stoping from id: %d\n" % worker_id)# 調(diào)用coord.request_stop()函數(shù)來(lái)通知其他線程停止。
            coord.request_stop()else:# 打印當(dāng)前線程的Id。print("Working on id: %d\n" % worker_id)# 暫停1秒time.sleep(1)# 聲明一個(gè)tf.train.Coordinator類來(lái)協(xié)同多個(gè)線程。
coord = tf.train.Coordinator()
# 聲明創(chuàng)建5個(gè)線程。
threads = [threading.Thread(target=MyLoop, args=(coord, i, )) for i in range(5)]
# 啟動(dòng)所有的線程
for t in threads:t.start()
# 等待所有線程退出
coord.join(threads)

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

?

?如何使用tf.QueueRunner和tf.Coordinator來(lái)管理多線程隊(duì)列操作。

#!/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來(lái)管理多線程隊(duì)列操作。
"""import tensorflow as tf# 聲明一個(gè)先進(jìn)先出的隊(duì)列，隊(duì)列中最多100個(gè)元素，類型為實(shí)數(shù)
queue = tf.FIFOQueue(100, "float")
# 定義隊(duì)列的入隊(duì)操作
enqueue_op = queue.enqueue([tf.random_normal([1])])# 使用tf.train.QueueRunner來(lái)創(chuàng)建多個(gè)線程運(yùn)行隊(duì)列的入隊(duì)操作。
# tf.train.QueueRunner的第一個(gè)參數(shù)給出了被操作的隊(duì)列，[enqueue_op] * 5
# 表示了需要啟動(dòng)5個(gè)線程，每個(gè)線程中運(yùn)行的是enqueue_op操作
qr = tf.train.QueueRunner(queue, [enqueue_op] * 5)# 將定義過(guò)的QueueRunner加入Tensorflow計(jì)算圖上指定的集合。
# tf.train.add_queue_runner函數(shù)沒有指定集合
# 則加入默認(rèn)集合tf.GraphKeys.QUEUE_RUNNERS。下面的函數(shù)就是將剛剛定義的
# qr加入默認(rèn)的tf.GraphKeys.QUEUE_RUNNER集合。
tf.train.add_queue_runner(qr)
# 定義出隊(duì)操作
out_tensor = queue.dequeue()with tf.Session() as sess:# 使用tf.train.Coordinator來(lái)協(xié)同啟動(dòng)的線程。coord = tf.train.Coordinator()# 使用tf.train.QueueRunner時(shí)，需要明確調(diào)用tf.train.start_queue_runners# 來(lái)啟動(dòng)所有線程。否則因?yàn)闆]有線程運(yùn)行入隊(duì)操作，當(dāng)調(diào)用出隊(duì)操作的時(shí)候，程序會(huì)一直# 等待入隊(duì)操作被運(yùn)行。tf.train.start_queue_runners函數(shù)會(huì)默認(rèn)啟動(dòng)# tf.GraphKeys.QUEUE_RUNNERS集合中所有的QueueRunner。因?yàn)檫@個(gè)函數(shù)值支持啟動(dòng)# 指定集合中的QueueRunner，所以一般來(lái)說(shuō)tf.train.add_queue_runner函數(shù)和# tf.trian.start_queue_runners函數(shù)會(huì)指定同一個(gè)集合。threads = tf.train.start_queue_runners(sess=sess, coord=coord)# 獲取隊(duì)列中的取值。for _ in range(3):print(sess.run(out_tensor)[0])# 使用tf.train.Coordinator來(lái)停止所有的線程
    coord.request_stop()coord.join(threads)

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

?

轉(zhuǎn)載于:https://www.cnblogs.com/lyjun/p/10350282.html

總結(jié)

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

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