這篇文章摘自我的博客, 歡迎大家沒事去逛逛~

這幾個(gè)月我開發(fā)了公司里的一個(gè)restful webservice，起初技術(shù)選型的時(shí)候是采用了flask框架。雖然flask是一個(gè)同步的框架，但是可以配合gevent或者其它方式運(yùn)行在異步的容器中(測試鏈接)，效果看上去也還可以，因此就采用了這種方式。

后面閱讀了tornado的源碼，也去了解了各種協(xié)程框架以及運(yùn)行的原理?？偢杏Xflask的這種同步方式編程不夠好，同時(shí)對于這種運(yùn)行在容器里的模式目前還缺乏了解。但至少現(xiàn)在對于tornado的運(yùn)行原理有了一定的了解，如果用tornado寫的話，是很可控的，而且可以保證運(yùn)行是高效的。因此就決定把原來基于flask的項(xiàng)目用tornado重構(gòu)了。

項(xiàng)目重構(gòu)的過程中遇到了一些問題，也學(xué)習(xí)了一些東西，這里做一個(gè)簡單的總結(jié)。

所有框架都要處理的一個(gè)接入層的事情就是：

url-mapping

項(xiàng)目初始化

參數(shù)解析

對于restful風(fēng)格的接口以及項(xiàng)目的初始化，每個(gè)框架都有自己的方式，在它們的文檔中都演示得特別清楚，所以關(guān)于這些我就不展開了。

關(guān)于參數(shù)解析，這里并不是指簡單地調(diào)用類似于get_argument這樣的方法去獲取數(shù)據(jù)。而是 如何從不可靠的client端傳來的數(shù)據(jù)中過濾掉服務(wù)器不關(guān)注的數(shù)據(jù)，同時(shí)對服務(wù)器關(guān)注的數(shù)據(jù)作一些更強(qiáng)的校驗(yàn)，這就是協(xié)議層的事情了。

使用谷歌的ProtocolBuffer是一個(gè)不錯(cuò)的方案，它有很不錯(cuò)的數(shù)據(jù)壓縮率，也支持目前大多數(shù)主流的開發(fā)語言。但在一些小型項(xiàng)目中，我還是更偏向于使用json的方式，它顯得更加靈活。但是對于json的話，如何作數(shù)據(jù)校驗(yàn)就是另外一個(gè)問題了。

在重構(gòu)前，我是通過python中的裝飾器來實(shí)現(xiàn)的這個(gè)功能:

class SomeHandlerInFlask(Resource):
    @util.deco({
        "key_x": (str, "form"),
        "key_y": (int, "form"),
        "key_z": (str, "url")
    })
    def post(self):
        # logic code
        pass

在裝飾器中分別從不同的地方，form或者url中獲取相應(yīng)的參數(shù)。如果取不到，則直接報(bào)錯(cuò)，邏輯也不會進(jìn)入到post函數(shù)中。

這是我基于flask這個(gè)框架自己總結(jié)出來的一套尚且還能看能用的參數(shù)解析方式，如果在每個(gè)函數(shù)中通過框架提供的get_argument來逐一獲取參數(shù)，則顯得太丑，而且每個(gè)接口所需要的數(shù)據(jù)是什么也不夠直觀。不過這種方式我自己還不是特別滿意，總感覺還是有點(diǎn)不太舒服，也說不清不舒服在哪里。那就干脆放棄它，使用別的方式吧。

后來我了解到了jsonschema這個(gè)東西，看了一下感覺與ProtocolBuffer很相似，只不過它是采用json的格式定義，正合我意(對于它我也有點(diǎn)吐槽，在數(shù)據(jù)庫層有提到)，每次數(shù)據(jù)進(jìn)來就對數(shù)據(jù)和schema作一次validate操作，再進(jìn)入業(yè)務(wù)邏輯層。

業(yè)務(wù)邏輯層的重構(gòu)其實(shí)改動的代碼并不多，把一些同步的操作改成異步的操作。就拿如何重構(gòu)某個(gè)接口來說吧，重構(gòu)前的代碼可能是這樣的:

def function_before_refactor(some_params):
    result_1 = sync_call_1(some_params)
    result_2 = sync_call_2(some_params)
    # some other processes
    return result

使用gen.coroutine重構(gòu)后:

from tornado import gen

@gen.coroutine
def function_after_refactor(some_params):
    # if you don"t want to refactor
    # just call it as it always be
    result_1 = sync_call_1(some_params)
    result_2 = yield async_call_2(some_params)
    # some other processes
    raise gen.Return(result)
    # python3及以上的版本不需要采用拋出異常的方式，直接return就可以了
    # return result

考慮到函數(shù)名根本不用改，重構(gòu)的過程非常容易：

函數(shù)用gen.coroutine包裝成協(xié)程

已經(jīng)重構(gòu)成異步方式的函數(shù)調(diào)用時(shí)添加yield關(guān)鍵字即可

函數(shù)返回采用raise gen.Return(result)的方式（僅限于Python 2.7）

因?yàn)槲夷壳安捎玫氖莗ython 2.7，所以在處理返回的時(shí)候要用拋出異常的方式，在這種方式下有一個(gè)點(diǎn)需要注意到，那就是與平常異常的處理的混用，不然會導(dǎo)致邏輯流執(zhí)行混亂：

from tornado import gen

@gen.coroutine
def function_after_refactor(some_params):
    try:
        # some logic code
        pass
    except Exception as e:
        if isinstance(e, gen.Return):
            # return the value raised by logic
            raise gen.Return(e.value)
        # more exception process

數(shù)據(jù)庫采用的是mongodb，在flask框架中采用了mongoengine作為數(shù)據(jù)庫層的orm，對于這個(gè)python-mongodb的orm產(chǎn)品，我個(gè)人并不是很喜歡(可能是因?yàn)槲伊?xí)慣了mongoose的工作方式)，這里面嵌套json的定義居然不能體現(xiàn)在schema中，需要分開定義兩個(gè)schema，然后再作引入的操作。比如（代碼只是用作演示，與項(xiàng)目無關(guān)）：

class Comment(EmbeddedDocument):
    content = StringField()
    # more comment details

class Page(Document):
    comments = ListField(EmbeddedDocumentField(Comment))
    # more page details

而在mongoose中就直觀多了:

var PageSchema = new Schema({
    title       :   {type : String, required : true},
    time        :   {type : Date, default : Date.now(), required : true},
    comments    :   [{
        content   :   {type : String}
        // more comment details
    }]
    // more page details
});

扯遠(yuǎn)了，在tornado的框架中，再使用mongoengine就不合適了，畢竟有著異步和同步的區(qū)別。那有什么比較好的python-mongodb的異步orm框架呢？搜了下，有一個(gè)叫做motorengine的東西，orm的使用方式和mongoengine基本一樣，但看它的star數(shù)實(shí)在不敢用呀。而且它處理異步的方式是使用回調(diào)，現(xiàn)在都是使用協(xié)程的年代了，想想還是算了吧。

最后找了個(gè)motor，感覺還不錯(cuò)，它有對目前大部分主流協(xié)程框架的支持，操作mongodb的方式與直接使用pymongo的方式差不多（畢竟都是基于pymongo的封裝嘛），但是就是沒有orm的驗(yàn)證層，那就自己再去另外搞一個(gè)簡化的orm層吧。(mongokit的orm方式看上去還不錯(cuò)，但貌似對協(xié)程框架的支持一般)。這里暫時(shí)先懶惰一下，還是采用了jsonschema。每次保存前都validate一下對象是否符合schema的定義。如果沒有類mongoose的python-mongodb異步框架，有時(shí)間就自己寫一個(gè)吧~

這里順帶吐槽一下jsonschema，簡直太瑣碎了，一個(gè)很短的文檔結(jié)構(gòu)定義，它會描述成好幾十行，我就不貼代碼了，有興趣的朋友可以戳這里http://jsonschema.net/玩玩。而且python中的jsonschema庫還不支持對于default關(guān)鍵字的操作，參見這個(gè)issue。

python中的測試框架有很多，只要選擇一個(gè)合適的能夠很方便與項(xiàng)目集成就好。我個(gè)人還是很喜歡unittest這個(gè)框架，小而精。我的這套測試方案也是基于unittest框架的。

# TestUserPostAccessComponents.py
class TestUserPostAccessComponents(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        # 定義在其它地方，具體細(xì)節(jié)就不展示了
        # 在setup中使用測試賬號獲取登陸態(tài)
        # 并把各種中間用得到的信息放在TestUserPostAccess類上
        setup(cls)

    @classmethod
    def tearDownClass(cls):
        pass

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_1_user_1_user_2_add_friend(self):
        pass

    def test_2_user_1_user_2_del_friend(self):
        pass

    def test_3_user_1_add_public_user_post(self):
        pass

    # more other components

最頂層的測試文件：

# run_test.py
# 各種import

def user_basic_post_access_test():
    tests = ["test_3_user_1_add_public_user_post",
             "test_5_user_2_as_a_stranger_can_access_public_user_post",
             "test_4_user_1_del_public_user_post",
             "test_6_user_1_add_private_user_post",
             "test_8_user_2_as_a_stranger_can_not_access_private_user_post",
             "test_9_user_1_self_can_access_private_user_post",
             "test_7_user_1_del_private_user_post"]
    return unittest.TestSuite(map(TestUserPostAccessComponents, tests))

def other_process_test():
    tests = [
        # compose a process by components by yourself
    ]
    return unittest.TestSuite(map(OtherTestCaseComponents, tests))

runner = unittest.TextTestRunner(verbosity=2)
runner.run(user_basic_post_access_test())
runner.run(other_process_test())

這套測試是基于 BDD (行為驅(qū)動)的測試方式，針對每一個(gè)邏輯模塊，定義一個(gè)components類，把所有子操作都定義成多帶帶的測試單元。這里面的測試單元可以是完全無序的，把邏輯有序化組織成測試用例的過程會在最外面通過TestSuit的方式組織起來。這里可能會有一些異議，因?yàn)橛行┤嗽谑褂眠@個(gè)測試類的時(shí)候是把它作為一個(gè)測試用例來組織的，當(dāng)然這些都是不同的使用方式。

這套測試方案中的每個(gè)component都是api級別的測試，并不是函數(shù)級別的測試（集成測試與單元測試），每個(gè)TestSuit都是完整的一個(gè)業(yè)務(wù)流程。這樣的好處在于 測試和項(xiàng)目完全解耦。測試代碼不用關(guān)心項(xiàng)目的代碼是同步還是異步的。就算項(xiàng)目重構(gòu)了，測試完全無感知，只要api沒變，就可以繼續(xù)工作。

當(dāng)然以上都是理想的狀態(tài)，因?yàn)樵趧傞_始寫這些測試的時(shí)候我還沒有總結(jié)到這些點(diǎn)，導(dǎo)致了一些耦合性的存在。比如說測試代碼中import了項(xiàng)目中的某個(gè)函數(shù)去獲取一些數(shù)據(jù)，用于檢查某個(gè)component的更新操作是否成功。在重構(gòu)的過程中，該函數(shù)被重構(gòu)成了協(xié)程。這樣一來，在測試代碼中就不能采用原來一樣的方式去調(diào)用了，也就是說測試代碼受到了框架同步與異步的影響，下一節(jié)我們就來談?wù)勍脚c異步的測試，以及對于這種問題的解決方案。

在tornado中，也提供了一套測試的功能，具體在tornado.testing這個(gè)模塊，看它源碼其實(shí)可以發(fā)現(xiàn)它也是基于unittest的一層封裝。
我心里一直有一個(gè)問題：unittest的執(zhí)行流程是同步的，既然這樣，它是怎么去測一個(gè)由gen.coroutine包裝的協(xié)程的呢，畢竟后者是異步的。
直到看了源碼，恍然大悟，原來是io_loop.run_sync這個(gè)函數(shù)的功勞，具體實(shí)現(xiàn)在gen_test這個(gè)裝飾器中，摘一部分源碼（對于tornado源碼不熟的同學(xué)可以先去看看tornado中的ioloop模塊的實(shí)現(xiàn)，看完會對這個(gè)部分有更深刻的理解）：

def gen_test(func=None, timeout=None):
    if timeout is None:
        timeout = get_async_test_timeout()

    def wrap(f):
        # Stack up several decorators to allow us to access the generator
        # object itself.  In the innermost wrapper, we capture the generator
        # and save it in an attribute of self.  Next, we run the wrapped
        # function through @gen.coroutine.  Finally, the coroutine is
        # wrapped again to make it synchronous with run_sync.
        #
        # This is a good case study arguing for either some sort of
        # extensibility in the gen decorators or cancellation support.
        @functools.wraps(f)
        def pre_coroutine(self, *args, **kwargs):
            result = f(self, *args, **kwargs)
            if isinstance(result, types.GeneratorType):
                self._test_generator = result
            else:
                self._test_generator = None
            return result

        coro = gen.coroutine(pre_coroutine)

        @functools.wraps(coro)
        def post_coroutine(self, *args, **kwargs):
            try:
                return self.io_loop.run_sync(
                    functools.partial(coro, self, *args, **kwargs),
                    timeout=timeout)
            except TimeoutError as e:
                # run_sync raises an error with an unhelpful traceback.
                # If we throw it back into the generator the stack trace
                # will be replaced by the point where the test is stopped.
                self._test_generator.throw(e)
                # In case the test contains an overly broad except clause,
                # we may get back here.  In this case re-raise the original
                # exception, which is better than nothing.
                raise
        return post_coroutine

    if func is not None:
        # Used like:
        #     @gen_test
        #     def f(self):
        #         pass
        return wrap(func)
    else:
        # Used like @gen_test(timeout=10)
        return wrap

在源碼中，先把某個(gè)測試單元封裝成一個(gè)協(xié)程，然后獲取當(dāng)前線程的ioloop對象，把協(xié)程拋給他去執(zhí)行，直到執(zhí)行完畢。這樣就完美地實(shí)現(xiàn)了異步到同步的過渡，滿足unittest測試框架的同步需求。
在具體的使用中只需要繼承tornado提供的AsyncTestCase類就行了，注意這里不是unittest.TestCase?？戳嗽创a也可以發(fā)現(xiàn)，前者就是繼承自后者的。

# This test uses coroutine style.
class MyTestCase(AsyncTestCase):
    @tornado.testing.gen_test
    def test_http_fetch(self):
        client = AsyncHTTPClient(self.io_loop)
        response = yield client.fetch("http://www.tornadoweb.org")
        # Test contents of response
        self.assertIn("FriendFeed", response.body)

回到上一節(jié)的問題，有了這種方式，就可以很容易地解決同步異步的問題了。如果測試用例中某一個(gè)函數(shù)已經(jīng)被項(xiàng)目重構(gòu)成了協(xié)程，只需要做以下三步：

把測試components的類改成繼承自AsyncTestCase

該測試單元使用gen_test裝飾（其它測試單元可以不用加，只需要改涉及到協(xié)程的測試單元就行）

調(diào)用協(xié)程的地方添加yield關(guān)鍵字

如果是api測試
測試中盡量不要調(diào)用任何項(xiàng)目中的代碼，它只專注于測試接口是否按照預(yù)期在工作，具體里面是怎么樣的不需要關(guān)心。這樣的話整套測試是完全獨(dú)立于項(xiàng)目而存在的，即使項(xiàng)目重構(gòu)，也可以不用作任何修改，無縫對接。

如果是單元測試
參考上一節(jié)的方案。

重構(gòu)是一個(gè)不斷優(yōu)化和學(xué)習(xí)的過程，在這個(gè)過程中我踩了一些坑，也爬出了一些坑，希望可以把我的這些總結(jié)分享給大家。歡迎大家跟我交流。對于文中的一些方案，也歡迎大家拍磚，歡迎有更多的做法可以一起探討學(xué)習(xí)。另外，對于這個(gè)項(xiàng)目的重構(gòu)，文章里面可能還少了一些更加直觀的性能測試，后面我會加上去，孝敬各位爺~