lwzm
12/11/2015 - 9:09 AM

tornado_threadpool_test.py

#!/usr/bin/env python3

import collections
import csv
import datetime
import functools
import ipaddress
import json
import logging
import sys
import time
import urllib.parse
import xmlrpc.client

import tornado.gen
import tornado.httpclient
import tornado.ioloop
import tornado.log
import tornado.util
import tornado.web

executor = tornado.concurrent.futures.ThreadPoolExecutor(4)


def sleep(seconds):
    time.sleep(seconds)
    return seconds * 1000


class Handler(tornado.web.RequestHandler):
    executor = tornado.concurrent.futures.ThreadPoolExecutor(4)

    @tornado.concurrent.run_on_executor
    def sleep(self, seconds=2):
        time.sleep(seconds)
        return seconds * 1000

    @tornado.gen.coroutine
    def get(self, seconds=1):
        seconds = float(seconds)
        #yield tornado.gen.sleep(2)
        #result = yield executor.submit(sleep, float(seconds))
        result = yield self.sleep(seconds)
        self.write("{} ok".format(result))


class MainHandler(tornado.web.RequestHandler):
    @tornado.gen.coroutine
    def get(self):
        self.write("ok")


handlers = [
    (r"/(.+)", Handler),
    (r"/",     Handler),
]

def listen(port):
    tornado.web.Application(
        handlers,
        static_path="static",
        template_path="templates",
        debug=__debug__,
    ).listen(port, xheaders=True)


def main():
    from tornado.options import define, options, parse_command_line
    define("port", 8888, type=int)
    parse_command_line()

    listen(options.port)

    io_loop = tornado.ioloop.IOLoop.instance()

    print(time.time())
    io_loop.start()


if __name__ == "__main__":
    main()