TensorFlow中tf.FIFOQueue提前关闭的原因及修复方案问询

首先，咱们来拆解你遇到的OutOfRangeError问题，核心原因和修复思路如下：

错误原因 #

你触发错误的关键在于队列线程启动时机不对，导致队列提前被关闭：

1. 你在初始化data_变量之前就启动了QueueRunner后台线程，线程启动后会立刻尝试执行q.enqueue(data2)操作。但此时data_还未初始化，计算data2会触发未初始化变量的错误，直接导致后台线程崩溃退出，同时队列被标记为"关闭"状态。
2. 第一次调用data_.initializer时，队列里还有你手动入队的初始值data0，所以能成功取出；但此时队列已经被关闭，后续任何对关闭队列的dequeue操作（比如第二次调用data_.initializer）都会直接抛出OutOfRangeError——哪怕队列里还有元素，关闭的队列也不允许再进行出队操作。

另外，你的循环逻辑也有小问题：当前代码里data_是一个变量，一旦初始化后值就固定了，循环里10次run(data2)打印的都是同一个值，这显然不是你想要的"随优化步骤更新缓冲区"的效果。

修复方案 #

针对这两个问题，我们可以分两步调整代码：

方案1：修正线程启动时机+保留Variable逻辑 #

如果你坚持要用Variable来保存出队结果（避免同一次run中多次出队），需要先初始化data_再启动后台线程，确保线程执行入队操作时data_已有有效值：

import tensorflow as tf
import numpy as np

VarType = tf.int32
data0 = np.array([1], dtype=np.int32)  # 直接用int类型，避免隐式类型转换问题
init = tf.placeholder(VarType, [1])
q = tf.FIFOQueue(capacity=1, shapes=[1], dtypes=VarType)
nq_init = q.enqueue(init)

# 定义变量保存出队结果，排除在全局初始化集合外
data_ = tf.Variable(q.dequeue(), trainable=False, collections=[])
data1 = data_ + 1
data2 = data_ * data1

# 定义入队操作，用tf.identity确保每次入队的是当前计算结果
enqueue_op = q.enqueue(tf.identity(data2))
qr = tf.train.QueueRunner(q, [enqueue_op])
tf.train.add_queue_runner(qr)

with tf.Session() as sess:
    sess.run(tf.global_variables_initializer())
    coord = tf.train.Coordinator()
    
    # 先手动入队初始值，初始化data_变量
    sess.run(nq_init, feed_dict={init: data0})
    sess.run(data_.initializer)
    
    # 等data_初始化完成后，再启动队列线程
    threads = tf.train.start_queue_runners(coord=coord)
    
    try:
        for n in range(10):
            current_data2 = sess.run(data2)
            print(current_data2)
            # 重新初始化data_，取出队列里的新值（由后台线程入队）
            sess.run(data_.initializer)
        
        # 现在第二次初始化就不会报错了
        sess.run(data_.initializer)
        print("第二次初始化后，data2的值：", sess.run(data2))
    except tf.errors.OutOfRangeError:
        print("队列已空")
    finally:
        coord.request_stop()
        coord.join(threads)
print('Done')

方案2：简化逻辑，不用Variable（更符合你的需求） #

其实你不需要用Variable来避免多次出队——直接用一个tensor保存出队结果，同一次run中多次引用这个tensor只会执行一次出队操作。这种方式更简洁，也更贴合你"缓冲区随步骤更新"的需求：

import tensorflow as tf
import numpy as np

VarType = tf.int32
data0 = np.array([1], dtype=np.int32)
q = tf.FIFOQueue(capacity=1, shapes=[1], dtypes=VarType)
# 初始入队操作
init_enqueue = q.enqueue(data0)

# 单次出队，基于这个结果计算后续值
data_dequeued = q.dequeue()
data1 = data_dequeued + 1
data2 = data_dequeued * data1
# 后台线程负责将计算结果重新入队
enqueue_op = q.enqueue(data2)

qr = tf.train.QueueRunner(q, [enqueue_op])
tf.train.add_queue_runner(qr)

with tf.Session() as sess:
    coord = tf.train.Coordinator()
    threads = tf.train.start_queue_runners(coord=coord)
    
    # 先执行初始入队
    sess.run(init_enqueue)
    
    try:
        for n in range(10):
            # 每次run都会自动取出队列里的新值，计算后后台线程会把结果入队
            current_data2 = sess.run(data2)
            print(current_data2)
            # 输出会是：2, 6, 42, 1806... 符合缓冲区更新的逻辑
    except tf.errors.OutOfRangeError:
        print("队列已空")
    finally:
        coord.request_stop()
        coord.join(threads)
print('Done')

这个版本的代码会每次循环都取出队列里的最新值，计算后再放回队列，完美实现你想要的"依赖缓冲区值的优化器循环"效果。

内容的提问来源于stack exchange，提问作者AOK