RandomShuffleQueue

Posted 2020-12-25 coolqiyu

一、几个函数

• RandomShuffleQueue类

__init__(self, capacity, min_after_dequeue,dtypes, shapes=None,names=None, seed=None, shared_name=None, name="random_shuffle_queue")

说明

queue = tf.RandomShuffleQueue(...)：创建一个queue，按随机顺序进行dequeue

RandomShuffleQueue有一定的容量限制capacity，支持多个生产者和消费者
RandomShuffleQueue中的每个元素是固定长度的tensor 元组，数据类型由dtypes定义，形状为shapes。如果shapes没有定义，那么不同的queue元素可能有不同的形状，此时就不能使用dqueue_many。如果shapes定义了，则所有的元素必须有相同的形状
min_after_dequeue决定queue在dequeue以后要保持的元素个数，如果没有足够的元素，就会block住dequeue的相关操作，直到有足够元素进来。当queue关闭，则这个参数被忽略

• enqueue(self, vals, name=None)

说明

enqueue_op = queue.enqueue(...) 创建enqueue元素到queue中的操作

如果操作执行时queue是满的，则会block住
vals是一个tensor或一个tensor的list/tuple，或者是一个字典，它相当于enqueue操作时的数据池
enqueue操作是要手动触发的，也就是不是说像一般的那种计算，会把enqueue作为依赖操作被执行

• queue.dequeue(self, name=None)

从queue中取出一个元素

• Coordinator类

__init__(self, clean_stop_exception_types=None)

协调线程的执行

• QueueRunner类

__init__(self, queue=None, enqueue_ops=None, close_op=None, cancel_op=None, queue_closed_exception_types=None,queue_runner_def=None, import_scope=None)

说明
qr = tf.train.QueueRunner(...) 为一个queue保持一系列enqueue操作，每个操作以一个线程执行
queue: a Queue
enqueue_ops: 一个enqueue ops列表
close_op: 指定关闭queue的操作
cancel_op：指定关闭以及取消挂起的enqueue ops的操作

• qr.create_threads(self, sess, coord=None, daemon=False, start=False)

为给定的sess创建多个线程以执行enqueue ops
start：如果为False，则需要手动调用 start()来启动

• start_queue_runners

start_queue_runners(sess=None, coord=None, daemon=True, start=True, collection=ops.GraphKeys.QUEUE_RUNNERS)

tf.train.start_queue_runners(...) 启动图中所有的queue runners，与add_queue_runner()配合使用
start: `False`只是创建线程，但是没有启动

二、实例 

 1 def example1():
 2 
 3 """
 4 最简单的例子，只使用enqueue和dequeue
 5 :return: 
 6 """
 7 example = tf.constant(2, "float32", [2, 2])
 8 # 创建一个queue
 9 # tf.RandomShuffleQueue(capacity,: queue的容量
10 # min_after_dequeue, : 保证queue中最少的个数
11 # dtypes,
12 # shapes=None,...)
13 queue = tf.RandomShuffleQueue(10, 0, "float32", shapes=[2, 2])
14 # 为queue添加enqueue操作
15 enqueue_op = queue.enqueue(example)
16 # 为queue添加dequeue操作
17 inputs = queue.dequeue()
18 with tf.Session() as sess:
19 sess.run(tf.global_variables_initializer())
20 sess.run(enqueue_op)
21 print(sess.run(inputs))

View Code

 1 def example2():
 2 """
 3 使用queue runner来管理多个enqueue线程，用coord来关闭线程
 4 :return: 
 5 """
 6 data = tf.constant(2, "float32", [2, 2])
 7 example = [data, data, data, data, data, data, data, data]
 8 queue = tf.RandomShuffleQueue(10, 0, "float32", shapes=[2, 2])
 9 enqueue_op = queue.enqueue(example)
10 
11 qr = tf.train.QueueRunner(queue, [enqueue_op] * 4)
12 coord = tf.train.Coordinator()
13 
14 inputs = queue.dequeue()
15 with tf.Session() as sess:
16 threads = qr.create_threads(sess, coord, start=True)
17 sess.run(tf.global_variables_initializer())
18 print(sess.run(inputs))
19 # 用coord来停止所有的enqueu线程
20 coord.request_stop()
21 coord.join(threads)

View Code