tensorflow多线程之QueueRunner的小总结

比眉伴天荒 2022-03-08 09:12 155阅读 0赞

上一篇总结了FIFOQueue可知，有两种情况会挂起，一旦挂起，下面再花团锦簇的代码也是白瞎，不会得到执行，为了不让系统挂起，TF采用QueueRunner来控制对Queue的访问（尤其是程序猿自己写的复杂访问操作，不仅仅是enqueue,dequeue等）。QueueRunner可以创建独立的线程来实现对Queue的独立访问。  
和其他语言一样，所谓的线程，就是一段代码的运行。这段代码通常就是一个函数过程。  
定义一个QueueRunner的基操如下：  
qr=tf.nn.QueueRunner(queue,enqueue\_ops)  
然后就可以用这个创建的QueueRunner创建并运行线程了  
my\_threads=qr.create\_threads(sess,start=True,coord=cord)  
注意上面的创建语句，有一个coord=cord。是指定线程的管理的协调器变量。有了这个变量，才可以终止线程（QueueRunner只管生孩子，不管养孩子），当然也可以不用协调器，那么线程会出现挂起、出错等异常。  
注意，经过验证，QueueRunner只能管理TF 操作符代表的线程，不能创建我们用def xxx():创建的函数作为线程。要使用函数这样的复杂操作作为线程，需要直接使用Threading类型。当然，Threading类型也需要协调器管理。  
演示代码：

import tensorflow as tf
    import time
    q = tf.FIFOQueue(3, "float")
    counter=tf.Variable(0.0)
    init = q.enqueue([counter])
    
    counter_add=tf.assign_add(counter,1.0)
    my_thread_enqueue=q.enqueue([counter])
    
    cord=tf.train.Coordinator()
    qr=tf.train.QueueRunner(q,enqueue_ops=[my_thread_enqueue,counter_add]*2)
    
    with tf.Session() as sess:
        sess.run(tf.global_variables_initializer())
        sess.run(init)
        my_threads=qr.create_threads(sess,start=True,coord=cord)
        for i in range(100):
            my_dequeue=sess.run(q.dequeue())
            print(my_dequeue)
        cord.request_stop()
        cord.join(my_threads)

PS：  
qr=tf.train.QueueRunner(q,enqueue\_ops=\[my\_thread\_enqueue,counter\_add\]\*2)  
乘以2表示创建两份列表内的线程，此处就是4个线程：两个my\_thread\_enqueue，两个counter\_add。  
my\_threads=qr.create\_threads(sess,start=True,coord=cord)  
此语句在创建线程后就立即开始运行，也可以start=False，然后  
tf.train.start\_queue\_runners(coord=coord)启动图中所有线程