tf多线程读取数据

# encoding:utf-8
__author__ &＃61; &＃39;HP&＃39;
import tensorflow as tf# 新建一个Session
with tf.Session() as sess:# 文件名listfilename &＃61; [&＃39;2.png&＃39;, &＃39;3.png&＃39;]# string_input_producer会产生一个文件名队列filename_queue &＃61; tf.train.string_input_producer(filename, shuffle&＃61;False, num_epochs&＃61;2)# reader从文件名队列中读数据。对应的方法是reader.readreader &＃61; tf.WholeFileReader()key, value &＃61; reader.read(filename_queue)# tf.train.string_input_producer定义了一个epoch变量&＃xff0c;要对它进行初始化
tf.local_variables_initializer().run()# 使用start_queue_runners之后&＃xff0c;才会开始填充队列threads &＃61; tf.train.start_queue_runners(sess&＃61;sess)i &＃61; 0while True:i &＃43;&＃61; 1# 获取图片数据并保存image_data &＃61; sess.run(value)with open(&＃39;test_%d.jpg&＃39; % i, &＃39;wb&＃39;) as f:f.write(image_data)

tf.FIFOQueue(capacity, dtypes, shapes&＃61;None, names&＃61;None ...)

import tensorflow as tftf.InteractiveSession()q &＃61; tf.FIFOQueue(2, "float") # 最多2个元素
init &＃61; q.enqueue_many(([0,0],)) # 初始化队列

x &＃61; q.dequeue() # get
y &＃61; x&＃43;1
q_inc &＃61; q.enqueue([y]) # put

init.run() ## 初始化队列[0, 0]
# print(x.eval()) # 0.0
# print(x.eval()) # 0.0

q_inc.run() ## get 0 &＃43;1 put 1, 队列变成 [0, 1]
q_inc.run() ## get 0 &＃43;1 put 1&＃xff0c;队列变成 [1, 1]
q_inc.run() ## get 1 &＃43;1 put 2&＃xff0c;队列变成 [1, 2]
print(x.eval()) ## get 1
print(x.eval()) # get 2.0
x.eval() # 阻塞

__author__ &＃61; &＃39;HP&＃39;
# coding utf-8
import tensorflow as tf
import numpy as np
import threading
import time# 线程中运行的程序&＃xff0c;这个程序每隔1秒判断是否需要停止并打印自己的ID。
def MyLoop(coord, worker_id):# 使用tf.Coordinator类提供的协同工具判断当前线程是否需要停止并打印自己的IDwhile not coord.should_stop():# 人为制造一个停止的条件if np.random.rand() <0.1:print(&＃39;Stoping from id: %d\n&＃39; % worker_id)# 调用coord.request_stop()函数来通知其他线程停止
coord.request_stop()else:# 打印当前线程的IDprint(&＃39;Working on id: %d\n&＃39; % worker_id)# 暂停1秒time.sleep(1)# 声明一个tf.train.Coordinator类来协同多个线程
coord &＃61; tf.train.Coordinator()
# 声明创建5个线程
threads &＃61; [threading.Thread(target&＃61;MyLoop, args&＃61;(coord, i, )) for i in range(5)]
# 启动所有的线程
for t in threads: t.start()
# 等待所有线程退出
coord.join(threads)

import tensorflow as tf# 声明一个先进先出的队列&＃xff0c;队列中最多n个元素&＃xff0c;类型
queue &＃61; tf .FIFOQueue(10, &＃39;float&＃39;)
# 定义队列的入队操作
enqueue_op &＃61; queue.enqueue([tf.random_normal([1])])# 使用 tf.train.QueueRunner来创建多个线程运行队列的入队操作
# tf.train.QueueRunner给出了被操作的队列&＃xff0c;[enqueue_op] * 5
# 表示了需要启动5个线程&＃xff0c;每个线程中运行的是enqueue_op操作
qr &＃61; tf.train.QueueRunner(queue, [enqueue_op] * 5)
# 将定义过的QueueRunner加入TensorFlow计算图上指定的集合
# tf.train.add_queue_runner函数没有指定集合&＃xff0c;
# 则加入默认集合tf.GraphKeys.QUEUE_RUNNERS。
# 下面的函数就是将刚刚定义的qr加入默认的tf.GraphKeys.QUEUE_RUNNERS结合
tf.train.add_queue_runner(qr)
# 定义出队操作
out_tensor &＃61; queue.dequeue()with tf.Session() as sess:# 使用tf.train.Coordinator来协同启动的线程coord &＃61; tf.train.Coordinator()# 使用tf.train.QueueRunner时&＃xff0c;需要明确调用tf.train.start_queue_runners# 来启动所有线程。否则因为没有线程运行入队操作&＃xff0c;当调用出队操作时&＃xff0c;程序一直等待# 入队操作被运行。tf.train.start_queue_runners函数会默认启动# tf.GraphKeys.QUEUE_RUNNERS中所有QueueRunner.因为这个函数只支持启动指定集合中的QueueRunner,# 所以一般来说tf.train.add_queue_runner函数和tf.train.start_queue_runners函数会指定同一个结合threads &＃61; tf.train.start_queue_runners(sess&＃61;sess, coord&＃61;coord)# 获取队列中的取值for _ in range(11): print(sess.run(out_tensor)[0])# 使用tf.train.Coordinator来停止所有线程
coord.request_stop()coord.join(threads)

import tensorflow as tf# 文件名队列
filename_queue &＃61; tf.train.string_input_producer(["xx1.csv", "xx2.csv"], shuffle&＃61;False)# reader对象 &＃xff08;文件阅读器&＃xff09;
reader &＃61; tf.TextLineReader()
key, value &＃61; reader.read(filename_queue)# 为数据设定默认格式&＃xff0c;如果出现空值&＃xff0c;就替换为这种格式的默认值。
# 注意格式必须一样&＃xff0c;因为输出的格式是统一的&＃xff0c;如下
# [array([ 4. , 0. , 34.322323, 1. ], dtype&＃61;float32)]
record_defaults &＃61; [[1.], [1.], [1.], [1.]]
col1, col2, col3, col4 &＃61; tf.decode_csv(value, record_defaults&＃61;record_defaults) # decode_csv
features &＃61; tf.concat([[col1], [col2], [col3], [col4]], 0)with tf.Session() as sess:# Start populating the filename queue.coord &＃61; tf.train.Coordinator()threads &＃61; tf.train.start_queue_runners(coord&＃61;coord) # 启动队列&＃xff0c;相当于是启动了多个线程&＃xff0c;并将coord传入每个线程for i in range(12):example &＃61; sess.run([features])print(example)coord.request_stop() #
coord.join(threads) # 等待结束