浅析TF中的多线程

川长思鸟来 2022-03-24 14:46 234阅读 0赞

TF在从文件读取大量数据的时候，由于文件读取的速度明显会低于内存中数据处理的速度，按照传统的方法，单流程的程序需要频繁等待文件的读取，然后再处理读取的数据，程序运行效率很慢，为了解决这个问题，TF也采用了多线程处理机制，但是TF的封装特点，导致和我们以前学习的其他语言的多线程处理方式大相径庭，较难理解。  
为了配合数据存取，TF在实现多线程的时候需要三个类互相配合：Queue，QueueRunner和Coordinator。他们的的分工如下：

### Queue是队列类 ###

和我们数据结构学习的队列具有相同的性质，负责存取数据，包含的操作常用的就两个：入队：enqueue;和出队：dequeue。Queue详细分还分为 先进先出队列：FIFOQueue和随机乱序队列:RandomShuffleQueue。常用的是第一个。队列可以单独使用，不一定非得在多线程中才能使用。  
下面代码实现一个存储整数类型数据，长度为2的队列，入栈两个数据，再出栈这两个数据：

import tensorflow as tf
    queue=tf.FIFOQueue(2,tf.int16)
    tf.InteractiveSession()
    queue.enqueue(1).run()
    queue.enqueue(2).run()
    
    print(queue.dequeue().eval())
    print(queue.dequeue().eval())

Queue还有一个方法enqueue\_many()：可以一次入队多个数据，常用来初始化队列。

### QueueRunner ###

这个类型有两个作用：  
1.定义有队列上有哪些操作，这些操作就是一个个的线程。这些操作定义以后不会自动运行。  
2.通过create\_threads方法创建线程，创建时候需要指定线程在哪一个会话（session）里面运行，由哪一个Coordinator（协调器）来协调管理，以免线程不受控，此时开始，定义的线程才开始像脱缰的野狗一样激活运行。  
这个类型返回的是一个列表，每一项列表内容记录了所定义线程的名称、状态等信息。  
线程运行的另外一种方法是tf.train.start\_queue\_runners()。这个也比较常见到。这个方法会全局运行已经定义的线程，但是需要配合tf.train.add\_queue\_runner（）才能使用，否则会出现错误提示：

WARNING:tensorflow:`tf.train.start_queue_runners()` was called when no queue runners were defined. You can safely remove the call
    to this deprecated function.

如果你看到的代码没有add\_queue\_runner，但是使用了start\_queue\_runners也能正常运行，那么一定是在你import的库文件中已经有了add\_queue\_runner，所以不必惊奇。

### Coordinator ###

这个类负责管理那些像脱缰的野狗一样运行的线程，尽量让这些线程可控并且安全的结束，毕竟线程一旦开始运行，其运行到哪一个程度，我们是很难监测的，Queue一旦满，线程就不能写入数据，一旦空，线程就不能出队数据，此时导致线程挂起或出错。  
Coordinator常用两个方法：  
request\_stop（）通知各个线程结束当前工作  
join（）等待线程都正常结束。  
下面我们举例来演示三者之间是如何写作的：

import tensorflow as tf
    q = tf.FIFOQueue(20,'float')
    counter = tf.Variable(0.0)  # 计数器
    increment_op = tf.assign_add(counter, tf.constant(1.0))   # 计数器加一
    enqueue_op = q.enqueue(counter)                           # 入队
     
    # 线程面向队列q，启动2个线程，每个线程中是[in,en]两个操作,可以下图
    qr = tf.train.QueueRunner(q,enqueue_ops=[increment_op,enqueue_op]*2)
     
    sess = tf.Session()
    sess.run(tf.global_variables_initializer())
     
    coord = tf.train.Coordinator()
    # 线程管理器启动线程，接收协调器管理
    enqueue_thread = qr.create_threads(sess,coord=coord,start=True)
     
    for i in range(20):
        print(sess.run(q.dequeue()))
     
    coord.request_stop()            # 向各个线程发终止信号
    coord.join(enqueue_thread)      # 等待各个线程成功结束

上面的代码定义了四个线程（VS Code抓图），列表最后一个线程为框架自动加上去的：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxMTUwNDYz_size_16_color_FFFFFF_t_70]  
该程序输出可能如下：

10.0
    19.0
    69.0
    125.0
    146.0
    165.0
    197.0
    228.0
    305.0
    322.0
    335.0
    347.0
    370.0
    386.0
    401.0
    416.0
    448.0
    471.0
    492.0
    518.0

这个输出结果和我们想象的不一样，为什么呢？我们原先的算法是希望计数器加一以后入队，那么出队的结果应该也是差不多从1到200（虽然有两个计数器），而不是每一个数据相差这么大，原因在于这四个线程的运行速度不一样，而且定义的四个线程是分别独立运行的。自加线程明显运行更快，而入队线程则慢一些。可能自加线程已经加到146了，入队线程才完成上一个入队操作。所以导致入队数据相差较大，没有体现出数据每次累计加1的期望效果。  
这也体现了各个线程之间是同步运行，而不是等待-》运行这样的异步方式。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxMTUwNDYz_size_16_color_FFFFFF_t_70]: /images/20220323/fbf00c87d74741b49918170ebcd3e792.png