Tornado简单教程

`iostream` 对非阻塞式的`socket`的封装以便于常见读写操作
为了在处理请求时实现对`socket`的**异步读写**，Tornado实现了`IOStream`类用来处理`socket`的异步读写。

`ioloop` 核心的I/O循环
Tornado为了实现**高并发和高性能**，使用了一个`IOLoop`事件循环来处理`socket`的读写事件，`IOLoop`事件循环是**基于Linux的`epoll`模型**，可以高效地响应网络事件，这是Tornado高效的基础保证。

pip install tornado

#! /usr/bin/python
# encoding:utf-8

# 导入Tornado模块
import tornado.ioloop #核心IO循环模块
import tornado.httpserver #异步非阻塞HTTP服务器模块
import tornado.web #Web框架模块
import tornado.options #解析终端参数模块

#从终端模块中导出define模块用于读取参数，导出options模块用于设置默认参数
from tornado.options import define, options

# 定义端口用于指定HTTP服务监听的端口
# 如果命令行中带有port同名参数则会称为全局tornado.options的属性，若没有则使用define定义。
define("port", type=int, default=8000, help="run on the given port")

# 创建请求处理器
# 当处理请求时会进行实例化并调用HTTP请求对应的方法
class IndexHandler(tornado.web.RequestHandler):
    # 定义get方法对HTTP的GET请求做出响应
    def get(self):
        # 从querystring查询字符串中获取id参数的值，若无则默认为0.
        id = self.get_argument("id", 0)
        # write方法将字符串写入HTTP响应
        self.write("hello world id = " + str(id))

# 创建路由表
urls = [(r"/", IndexHandler),]

# 定义服务器
def main():
    # 解析命令行参数
    tornado.options.parse_command_line()
    # 创建应用实例
    app = tornado.web.Application(urls)
    # 监听端口
    app.listen(options.port)
    # 创建IOLoop实例并启动
    tornado.ioloop.IOLoop.current().start()

# 应用运行入口，解析命令行参数
if __name__ == "__main__":
    # 启动服务器
    main()

# 原型
def __init__(self, handlers=None, default_host="", transforms=None, wsgi=False, **settings):

server.listen(options.port)

from tornado.options import define, options

define("port", default=8888, help="run on the given port", type=int)

tornado.ioloop.IOLoop.current().start()

# 加载Tornado核心IO事件循环模块
import tornado.ioloop

# 默认Tornado的ioloop实例
tornado.ioloop.IOLoop.current()

# 创建应用实例
app = tornado.web.Application(urls)
# 监听端口
app.listen(options.port)

urls = [(r"/", MainHandler),]

# 当处理请求时会进行实例化并调用HTTP请求对应的方法
class MainHandler(tornado.web.RequestHandler):
    # 定义get方法对HTTP的GET请求做出响应
    def get(self):
        # 从querystring查询字符串中获取id参数的值，若无则默认为0.
        id = self.get_argument("id", 0)
        # write方法将字符串写入HTTP响应
        self.write("hello world id = " + id)

# 当处理请求时会进行实例化并调用HTTP请求对应的方法
from tornado.httpclient import HTTPClient

def synchronous_fetch(url):
    http_client = HTTPClient()
    response = http_client.fetch(url)
    return response.body

from tornado.httpclient import AsyncHTTPClient

def asynchronous_fetch(url, callback):
    http_client = AsyncHTTPClient()
    def handle_response(response):
        callback(response.body)
    http_client.fetch(url, callback=handle_response)

from tornado.concurrent import Future

def async_fetch_future(url):
    http_client = AsyncHTTPClient()
    my_future = Future()
    fetch_future = http_client.fetch(url)
    fetch_future.add_done_callback(
        lambda f: my_future.set_result(f.result()))
    return my_future

from tornado import gen

@gen.coroutine
def fetch_coroutine(url):
    http_client = AsyncHTTPClient()
    response = yield http_client.fetch(url)
    # 在 Python 3.3 之前的版本中, 从生成器函数
    # 返回一个值是不允许的,你必须用
    #   raise gen.Return(response.body)来代替
    return response.body

async def fetch_coroutine(url):
    http_client = AsyncHTTPClient()
    response = await http_client.fetch(url)
    return response.body

@gen.coroutine
def divide(x, y):
    return x / y

@gen.coroutine
def good_call():
    # yield will unwrap the Future returned by divide() and raise
    # the exception.
    yield divide(1, 0)

# The IOLoop will catch the exception and print a stack trace in
# the logs. Note that this doesn't look like a normal call, since
# we pass the function object to be called by the IOLoop.
IOLoop.current().spawn_callback(divide, 1, 0)

@gen.coroutine
def call_task():
    # Note that there are no parens on some_function.
    # This will be translated by Task into
    #   some_function(other_args, callback=callback)
    yield gen.Task(some_function, other_args)

thread_pool = ThreadPoolExecutor(4)

@gen.coroutine
def call_blocking():
    yield thread_pool.submit(blocking_func, args)

@gen.coroutine
def get(self):
    fetch_future = self.fetch_next_chunk()
    while True:
        chunk = yield fetch_future
        if chunk is None: break
        self.write(chunk)
        fetch_future = self.fetch_next_chunk()
        yield self.flush()

import motor
db = motor.MotorClient().test

@gen.coroutine
def loop_example(collection):
    cursor = db.collection.find()
    while (yield cursor.fetch_next):
        doc = cursor.next_object()

@gen.coroutine
def minute_loop():
    while True:
        yield do_something()
        yield gen.sleep(60)

# Coroutines that loop forever are generally started with
# spawn_callback().
IOLoop.current().spawn_callback(minute_loop)

@gen.coroutine
def minute_loop2():
    while True:
        nxt = gen.sleep(60)   # Start the clock.
        yield do_something()  # Run while the clock is ticking.
        yield nxt             # Wait for the timer to run out.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time     : 2023/5/29 19:19
# @Author   : 李泽雄
# @BoKeYuan : 小家电维修
# @File     : tornado_sql.py
# @Version  : Python 3.10.10
# @Project  : python3
# @Software : PyCharm

""" Mysql数据库操作 """

#import pymysql as MySQLdb
import logging
import pymysql
pymysql.install_as_MySQLdb()

logger = logging.getLogger(__name__)


class MysqlServer(object):
    """连接Mysql数据服务器 """

    def __init__(self, db_config):
        try:
            self._db_config = db_config
            self._conn = self.__get_conn()
            self._cursor = self._conn.cursor()
            logger.info(u"connected the db")
        except Exception:
            self.close()
            logger.exception(u"connect db failed!")

    def __get_conn(self):
        db_config = self._db_config
        connection = pymysql.connect(host=db_config['HOST'], port=db_config['PORT'], user=db_config['USERNAME'],
                                     passwd=db_config['PASSWORD'], db=db_config['DB'], charset="utf8")
        connection.ping(True)
        return connection

    def ensure_cursor(self):
        if not self._cursor:
            if not self._conn:
                self._conn = self.__get_conn()
            self._cursor = self._conn.cursor()

    def run_sql(self, sql):
        self.ensure_cursor()
        self._cursor.execute(sql)
        #commit只对innodb生效，不加commit的话，修改数据库记录的操作不会生效。而如果是myisam引擎的话，不需要commit即可生效
        self._conn.commit()
        return self._cursor.fetchall()

    def execute_sql(self, sql):
        self.ensure_cursor()
        self._cursor.execute(sql)
        self._conn.commit()

    def run_sql_fetchone(self, sql):
        self.ensure_cursor()
        self._cursor.execute(sql)
        return self._cursor.fetchone()

    def close(self):
        if self._cursor:
            self._cursor.close()
        if self._conn:
            self._conn.close()
        logger.info(u"closed the db connection")