tornado的简单使用

功能描述

实现一个类似与令牌桶的服务，包含一个定时任务，生成令牌，对外提供一个接口用于获取令牌，每获取一次，桶内令牌相应减少

实现方案

采用redis+tornado的方式实现，代码很简单，不过这里写这个主要是为了说明tornado的简单使用，同时介绍了一个tornado下的高效的redis异步库的使用

目录和文件结构

Makefile
setup.py
version.py
.bumpversion.cfg
src/
__init__.py
ratecontrol/
__init__.py
server/
__init__.py
app.py
start.py
data_client.py
urls.py

Makefile内容如下

all:
@echo "do nothing"

clean:
rm -f `find . -type f -name '*.py[co]' `
rm -fr */*.egg-info build dist

build: clean
python setup.py build_py bdist_wheel
cp Makefile dist

deploy:
pip install *.whl -U

release-major:
python setup.py release major

release-minor:
python setup.py release minor

release-patch:
python setup.py release patch

.PHONY : all clean build release-major release-minor release-patch

setup.py内容如下

# -*- coding: utf-8 -*-
import sys
import os
import setuptools
from version import __VERSION__

def _setup():
setuptools.setup(
name='ratecontrol',
version=__VERSION__,
description='rate control',
author='',
author_email='',
url='',
install_requires=['tornado==4.2','yatoredis'],
packages=['ratecontrol', 'ratecontrol.server'],
package_dir={'': 'src'},
entry_points={
'console_scripts': [
'ratecontrol-start=ratecontrol.server.start:ratecontrol_start'
]
},
classifiers=[
'Development Status :: 4 - Beta Development Status',
'Environment :: Console',
'Topic :: Utilities',
],
)

def main():
if len(sys.argv) > 1:
if sys.argv[1] == 'publish':
os.system('make publish')
sys.exit()
elif sys.argv[1] == 'release':
if len(sys.argv) < 3:
type_ = 'patch'
else:
type_ = sys.argv[2]
assert type_ in ('major', 'minor', 'patch')

os.system('bumpversion --current-version {} {}'
.format(__VERSION__, type_))
sys.exit()

_setup()

if __name__ == '__main__':
main()

.bumpversion.cfg

[bumpversion]
message = VERSION_TAG, Bumped: {current_version} -> {new_version}
commit = True
tag = True
current_version = 0.0.1
[bumpversion:file:version.py]

version.py

__VERSION__ = "0.0.1"

功能代码

data_client.py

这个文件主要实现了对redis内数据的更新，采用了yatoredis这个异步redis库，具体代码贴下

#coding=utf-8
from tornado import gen
from tornado.gen import Return
from toredis.client import Client as RedisClient
from toredis.pool import ClientPool as RedisClientPool
import logging

LOG = logging.getLogger(__name__)

class DataSource(object):

def __init__(self):
pass

def connect(self, host='localhost', port=6379):
self._redis_pool = RedisClientPool(
20,
host=host,
port=port)
@property
def redis_client(self):
return self._redis_pool.client

@gen.coroutine
def init_token(self, token_name, max_num):
yield gen.Task(self.redis_client.set, token_name, max_num)

@gen.coroutine
def get_token(self, token_name, token_num):
value = yield gen.Task(self.redis_client.decrby, token_name, token_num)
if None != value:
if value >= 0:
raise Return(True)
else:
yield gen.Task(self.redis_client.incrby, token_name, token_num)
raise Return(False)

@gen.coroutine
def incrby_token(self, token_name, max_num, token_num):
value = yield gen.Task(self.redis_client.get, token_name)
if int(value) < max_num:
incr = min(token_num, max_num-int(value))
yield gen.Task(self.redis_client.incrby, token_name, incr)

app.py

这个
e263
文件实现了对外的http接口

#coding=utf-8
import logging
from data_client import DataSource
from tornado import gen, web

_data_source = DataSource()

def get_data_source():
return _data_source

class RateControlHandler(tornado.web.RequestHandler):

@gen.coroutine
def get(self):
datas = get_data_source()
token_name = self.get_argument('token', '')
token_need = self.get_argument('need',1)
flag = False
if len(token_name) > 0:
flag = yield datas.get_token(token_name, token_need)
self.write({'status':flag})

class HealthCheckHandler(tornado.web.RequestHandler):

def get(self, *args, **kargs):
self.write('working')

def head(self, *args, **kargs):
self.write('working')

class HeatbeatHandler(tornado.web.RequestHandler):

def get(self, *args, **kargs):
self.write('ready')

def head(self, *args, **kargs):
self.write('ready')

urls.py

#coding=utf-8
from app import RateControlHandler
from app import HealthCheckHandler
from app import HeatbeatHandler

urls = [
(r'/api/rateicontrol/token/get', RateControlHandler),
(r'/healthcheck.html', HealthCheckHandler),
(r'/heartbeat.html', HeatbeatHandler),
]

start.py

这个文件包含了启动命令并且启动了一个周期任务来生成令牌

#coding=utf-8
from tornado.options import define, options, parse_command_line
from urls import urls
import logging
from tornado.web import Application as TornadoWebApplication
from app import get_data_source
import tornado.ioloop
from tornado.log import enable_pretty_logging
from tornado import gen

define("port", default=8880, help="run on the given port", type=int)
define("debug", default=False, help="run in debug mode")
define("redis_host", default="localhost", help="redis host ip")
define("redis_port", default=6379, help="redis port")
define("limit_time", default=3000, help="every # seconds incr the limit token", type=int)
define("token_max_limit", default=30000, help="the token max limit once send")
define("token_each_incr", default=3000, help="the token each incr")

_TOKEN_LIMIT = 'test_token'

def token_init():
datas = get_data_source()
datas.init_token(_TOKEN_LIMIT, options.token_max_limit)

@gen.coroutine
def token_incr():
datas = get_data_source()
yield datas.incrby_token(
_TOKEN_LIMIT, options.token_max_limit,
options.token_each_incr)

def ratecontrol_start():
enable_pretty_logging()
parse_command_line()
app = TornadoWebApplication(urls, debug=options.debug)
app.listen(options.port)
datas = get_data_source()
datas.connect(options.redis_host, options.redis_port)
token_init()
tornado.ioloop.PeriodicCallback(token_incr,options.limit_time).start()
tornado.ioloop.IOLoop.instance().start()

一点改进

tornado是可以在单个端口启动多进程的，对start.py的中ratecontrol_start函数可以按如下改写：

def ratecontrol_start():
enable_pretty_logging()
parse_command_line()
app = TornadoWebApplication(urls, debug=options.debug)
server = tornado.httpserver.HTTPServer(app)
server.bind(options.port)
server.start(2)
datas = get_data_source()
datas.connect(options.redis_host, options.redis_port)
io_loop = tornado.ioloop.IOLoop.instance()
io_loop.start()

一些依赖

部署时需要依赖wheel和bumpversion工具，可用pip安装