参考书
《TensorFlow：实战Google深度学习框架》（第2版）
对于队列，修改队列状态的操作主要有Enqueue、EnqueueMany和Dequeue。以下程序展示了如何使用这些函数来操作一个队列。
1#!/usr/bin/env python 2# -*- coding: UTF-8 -*- 3# coding=utf-8 4 5""" 6@author: Li Tian 7@contact: 694317828@qq.com 8@software: pycharm 9@file: queue_operate.py 10@time: 2019/1/31 21:32 11@desc: 操作一个队列 12""" 13 14import tensorflow as tf 15 16# 创建一个先进先出的队列，指定队列中最多可以保存两个元素，并指定类型为整数 17q = tf.FIFOQueue(2, "int32") 18# 使用enqueue_many函数来初始化队列中的元素。和变量初始化类似，在使用队列之前需要明确的调用这个初始化过程。 19init = q.enqueue_many(([0, 10],)) 20# 使用Dequeue函数将队列中的第一个元素出队列。这个元素的值将被存在变量x中 21x = q.dequeue() 22# 将得到的值+1 23y = x + 1 24# 将+1后的值再重新加入队列。 25q_inc = q.enqueue([y]) 26 27with tf.Session() as sess: 28 # 运行初始化队列的操作 29 init.run() 30 for _ in range(5): 31 # 运行q_inc将执行数据出队列、出队的元素+1、重新加入队列的整个过程。 32 v, _ = sess.run([x, q_inc]) 33 # 打印出队元素的取值 34 print(v)
运行结果：
tf.Coordinator主要用于协同多个线程一起停止，以下程序展示了如何使用tf.Coordinator。
1#!/usr/bin/env python 2# -*- coding: UTF-8 -*- 3# coding=utf-8 4 5""" 6@author: Li Tian 7@contact: 694317828@qq.com 8@software: pycharm 9@file: coordinator_test1.py 10@time: 2019/2/2 21:35 11@desc: tf.Coordinator主要用于协同多个线程一起停止，以下程序展示了如何使用tf.Coordinator 12""" 13 14import tensorflow as tf 15import numpy as np 16import threading 17import time 18 19 20# 线程中运行的程序，这个程序每隔1秒判断是否需要停止并打印自己的ID。 21def MyLoop(coord, worker_id): 22 # 使用tf.Coordinator类提供的协同工具判断当前线程是否需要停止 23 while not coord.should_stop(): 24 # 随机停止所有的线程。 25 if np.random.rand() < 0.1: 26 print("Stoping from id: %d\n" % worker_id) 27 # 调用coord.request_stop()函数来通知其他线程停止。 28 coord.request_stop() 29 else: 30 # 打印当前线程的Id。 31 print("Working on id: %d\n" % worker_id) 32 # 暂停1秒 33 time.sleep(1) 34 35 36# 声明一个tf.train.Coordinator类来协同多个线程。 37coord = tf.train.Coordinator() 38# 声明创建5个线程。 39threads = [threading.Thread(target=MyLoop, args=(coord, i, )) for i in range(5)] 40# 启动所有的线程 41for t in threads: 42 t.start() 43# 等待所有线程退出 44coord.join(threads)
运行结果：
如何使用tf.QueueRunner和tf.Coordinator来管理多线程队列操作。
1#!/usr/bin/env python 2# -*- coding: UTF-8 -*- 3# coding=utf-8 4 5""" 6@author: Li Tian 7@contact: 694317828@qq.com 8@software: pycharm 9@file: queuerunner_test1.py 10@time: 2019/2/3 12:31 11@desc: 如何使用tf.QueueRunner和tf.Coordinator来管理多线程队列操作。 12""" 13 14import tensorflow as tf 15 16# 声明一个先进先出的队列，队列中最多100个元素，类型为实数 17queue = tf.FIFOQueue(100, "float") 18# 定义队列的入队操作 19enqueue_op = queue.enqueue([tf.random_normal([1])]) 20 21# 使用tf.train.QueueRunner来创建多个线程运行队列的入队操作。 22# tf.train.QueueRunner的第一个参数给出了被操作的队列，[enqueue_op] * 5 23# 表示了需要启动5个线程，每个线程中运行的是enqueue_op操作 24qr = tf.train.QueueRunner(queue, [enqueue_op] * 5) 25 26# 将定义过的QueueRunner加入Tensorflow计算图上指定的集合。 27# tf.train.add_queue_runner函数没有指定集合 28# 则加入默认集合tf.GraphKeys.QUEUE_RUNNERS。下面的函数就是将刚刚定义的 29# qr加入默认的tf.GraphKeys.QUEUE_RUNNER集合。 30tf.train.add_queue_runner(qr) 31# 定义出队操作 32out_tensor = queue.dequeue() 33 34with tf.Session() as sess: 35 # 使用tf.train.Coordinator来协同启动的线程。 36 coord = tf.train.Coordinator() 37 # 使用tf.train.QueueRunner时，需要明确调用tf.train.start_queue_runners 38 # 来启动所有线程。否则因为没有线程运行入队操作，当调用出队操作的时候，程序会一直 39 # 等待入队操作被运行。tf.train.start_queue_runners函数会默认启动 40 # tf.GraphKeys.QUEUE_RUNNERS集合中所有的QueueRunner。因为这个函数值支持启动 41 # 指定集合中的QueueRunner，所以一般来说tf.train.add_queue_runner函数和 42 # tf.trian.start_queue_runners函数会指定同一个集合。 43 threads = tf.train.start_queue_runners(sess=sess, coord=coord) 44 # 获取队列中的取值。 45 for _ in range(3): 46 print(sess.run(out_tensor)[0]) 47 48 # 使用tf.train.Coordinator来停止所有的线程 49 coord.request_stop() 50 coord.join(threads)
运行结果：
