多线程-使用QueueRunner创建线程.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import tensorflow as tf"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 创建先进先出队列"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "queue = tf.FIFOQueue(100, \"float\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 入队操作，每次入队10个随机数值"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "enqueue = queue.enqueue([tf.random_normal([10])])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 使用QueueRunner创建10个线程进行队列入队操作\n",
    "#### 构造函数原型为_init_(self,queue,enqueue_ops,close_op,cancel_op,queue_closed_exception_types,queue_runner_def,import_scope)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "qr = tf.train.QueueRunner(queue, [enqueue] *10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 将定义过的QueueRuner加入到计算图的GraphKeys.QUEUE_RUNERS集合\n",
    "#### 函数原型add_queue_runner(qr,collection)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "tf.train.add_queue_runner(qr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 定义出队列操作"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "out_tensor = queue.dequeue()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[ 0.37119257 -0.7914396   0.32212585 -0.39406797 -0.7476919   0.5043666\n",
      "  0.08747364  0.541501    1.6405246   1.6201361 ]\n",
      "[ 0.54538435  0.2617037  -0.61257315  1.1765558   0.82286483 -0.8459544\n",
      "  0.09380955  0.5497491   0.4119415  -0.8576114 ]\n",
      "[-0.6268246   2.1273725  -1.9582455  -1.178697    0.5788816  -0.5880628\n",
      " -0.07411001 -0.66815364  0.6040597   0.8875204 ]\n",
      "[ 0.52197343 -0.31877983  0.94146556  0.51307297  1.8043536   0.24735409\n",
      " -0.5428715  -0.29021838  1.2911596   1.6328304 ]\n",
      "[-0.33352298 -0.84120893 -1.2129366  -0.33421662  1.0810916  -1.5639403\n",
      "  1.3628043  -0.8535187  -0.14825042  1.0816467 ]\n",
      "[-1.7872488   1.8904539  -0.25989455 -1.9107425   0.81714916 -0.56749994\n",
      " -2.518455   -0.61357534 -1.1356345  -0.77926147]\n",
      "[ 0.27107093  0.1798657  -0.2660226   1.3192132  -0.06427517 -0.600286\n",
      " -0.32701522  0.04854715  1.0111864   1.1215131 ]\n",
      "[ 1.1265213  -1.2268777   0.10607505  1.6806127   0.8483367  -0.5747743\n",
      "  0.6942945   0.66281915  1.7504295  -0.6820501 ]\n",
      "[-0.46348184 -0.21478797 -0.1113574  -0.97244924  0.3314631   0.20239718\n",
      "  0.2521348  -0.2355727   0.7421786  -0.10324737]\n",
      "[ 0.17523997  1.003546   -1.1614022  -0.234986   -0.10573537  2.0712724\n",
      " -1.3742027  -1.1110497  -1.3485744  -1.4594082 ]\n"
     ]
    }
   ],
   "source": [
    "with tf.Session() as sess:\n",
    "    # 使用Coordinator来协同启动的线程\n",
    "    coordinator = tf.train.Coordinator()\n",
    "\n",
    "    # 调用start_queue_runners()函数来启动所有线程，并通过参数coord指定\n",
    "    # 一个Coordinator来处理线程同步终止\n",
    "    # 函数原型start_queue_runners(sess,coord,daemon,start,collection)\n",
    "    threads = tf.train.start_queue_runners(sess=sess, coord=coordinator)\n",
    "\n",
    "    #打印全部的100个结果\n",
    "    for i in range(10):\n",
    "        print(sess.run(out_tensor))\n",
    "\n",
    "    #也可以这样定义打印的形式打印每个入队操作的第一个的结果\n",
    "    #for i in range(10):print(sess.run(out_tensor)[0])\n",
    "\n",
    "    coordinator.request_stop()\n",
    "    coordinator.join(threads)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [conda root]",
   "language": "python",
   "name": "conda-root-py"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}