import tensorflow as tf

def simple_shuffle_batch(source, capacity, batch_size=10):

    # Create a random shuffle queue.

    queue = tf.RandomShuffleQueue(capacity=capacity,

                                  min_after_dequeue=int(0.9*capacity),

                                  shapes=source.shape, dtypes=source.dtype)

    # Create an op to enqueue one item.

    enqueue = queue.enqueue(source)

    # Create a queue runner that, when started, will launch 4 threads applying

    # that enqueue op.

    num_threads = 4

    qr = tf.train.QueueRunner(queue, [enqueue] * num_threads)

    # Register the queue runner so it can be found and started by

    # <a href="../../api_docs/python/tf/train/start_queue_runners"><code>tf.train.start_queue_runners</code></a> later (the threads are not launched yet).

    tf.train.add_queue_runner(qr)

    # Create an op to dequeue a batch

    return queue.dequeue_many(batch_size)

# create a dataset that counts from 0 to 99

input = tf.constant(list(range(100)))

input = tf.data.Dataset.from_tensor_slices(input)

input = input.make_one_shot_iterator().get_next()

# Create a slightly shuffled batch from the sorted elements

get_batch = simple_shuffle_batch(input, capacity=20)

# `MonitoredSession` will start and manage the `QueueRunner` threads.

with tf.train.MonitoredSession() as sess:

    # Since the `QueueRunners` have been started, data is available in the

    # queue, so the `sess.run(get_batch)` call will not hang.

    while not sess.should_stop():

        print(sess.run(get_batch))

import tensorflow as tf

# Using Python's threading library.

import threading

import time

batch_size = 10

thread_num = 3

print("-" * 50)

def MyLoop(coord, id):

    step = 0

    while not coord.should_stop():

        step += 1

        print("thread id: %02d, step: %02d, ...do something..." %(id, step))

        time.sleep(0.01)

        if step >= 5:

            coord.request_stop()

# Main thread: create a coordinator.

coord = tf.train.Coordinator()

# Create thread_num threads that run 'MyLoop()'

threads = [threading.Thread(target=MyLoop, args=(coord,i)) for i in range(thread_num)]

# Start the threads and wait for all of them to stop.

for t in threads:

    t.start()

coord.join(threads)

print("-" * 50)

# create a dataset that counts from 0 to 99

example = tf.constant(list(range(100)))

example = tf.data.Dataset.from_tensor_slices(example)

example = example.make_one_shot_iterator().get_next()

# Create a queue, and an op that enqueues examples one at a time in the queue.

queue = tf.RandomShuffleQueue(capacity=20,

                              min_after_dequeue=int(0.9*20),

                              shapes=example.shape,

                              dtypes=example.dtype)

enqueue_op = queue.enqueue(example)

# Create a training graph that starts by dequeueing a batch of examples.

inputs = queue.dequeue_many(batch_size)

train_op = inputs # ...use 'inputs' to build the training part of the graph...

# Create a queue runner that will run thread_num threads in parallel to enqueue examples.

qr = tf.train.QueueRunner(queue, [enqueue_op] * thread_num)

# Launch the graph.

sess = tf.Session()

# Create a coordinator, launch the queue runner threads.

coord = tf.train.Coordinator()

enqueue_threads = qr.create_threads(sess, coord=coord, start=True)

# Run the training loop, controlling termination with the coordinator.

try:

    for step in range(1000000):

        if coord.should_stop():

            break

        y = sess.run(train_op)

        print(step, ",  y =", y)

except Exception as e:

    # Report exceptions to the coordinator.

    coord.request_stop(e)

finally:

    # Terminate as usual. It is safe to call `coord.request_stop()` twice.

    coord.request_stop()

    coord.join(threads)