创建队列
-
tf.FIFOQueue(capacity, dtypes) 先进先出队列，按顺序出队列
Q = tf.FIFOQueue(1000, tf.float32)
-
tf.RandomShuffleQueue(capacity, dtypes) 随机出队列
队列的方法和属性
Q.enqueue(vals)Q.enqueue_many(vals)
# 用逗号隔开指定是列表，防止与Tensor混淆
in_q_many = Q.enqueue_many([[1.0, 2.0, 3.0], ])
Q.dequeue()Q.dequeue_many(n)-
Q.size(). 计算此队列中的元素数量
队列管理器，开启多线程
-
tf.train.QueueRunner(queue=None, enqueue_ops=None)
queue：A Queue
enqueue_ops：添加线程的队列操作列表，[]*2,指定两个线程
# 定义子线程执行哪些操作
threads = tf.train.QueueRunner(Q, [in_q_many] * 2)
-
threads.create_threads(self, sess, coord=None, daemon=False, start=False)
创建并开启子线程
# start=True 自动开启线程，coord为子线程协调员
threads.create_threads(sess, coord=coord, start=True)
-
tf.train.Coordinator()
定义子线程协调员: 作用：管理子线程，防止主线程结束，子线程还没有退出
coord = tf.train.Coordinator()
完整代码
import tensorflow as tf
# 1. 创建队列
Q = tf.FIFOQueue(1000, tf.float32)
# 2. 定义操作 +1操作
var = tf.Variable(0.0)
data = tf.assign_add(var, tf.constant(1.0))
in_q = Q.enqueue(data)
# 3. 定义子线程执行哪些操作
threads = tf.train.QueueRunner(Q, [in_q] * 2)
init_var = tf.global_variables_initializer()
with tf.Session() as sess:
sess.run(init_var)
# 定义子线程协调员
coord = tf.train.Coordinator()
# 创建并开启子线程
threads.create_threads(sess, coord=coord, start=True)
for i in range(1000):
print(sess.run(Q.dequeue()))
# 结束子线程
coord.request_stop()
coord.join(threads)
