基于Sanic的微服務(wù)基礎(chǔ)架構(gòu)

seasonley 發(fā)布于2019-07-30 15:12 / 880人閱讀

摘要：在中，官方的異步協(xié)程庫(kù)正式成為標(biāo)準(zhǔn)。本項(xiàng)目就是以為基礎(chǔ)搭建的微服務(wù)框架。使用做單元測(cè)試，并且使用來(lái)避免訪問(wèn)其他微服務(wù)。跟蹤每一個(gè)請(qǐng)求，記錄請(qǐng)求所經(jīng)過(guò)的每一個(gè)微服務(wù)，以鏈條的方式串聯(lián)起來(lái)，對(duì)分析微服務(wù)的性能瓶頸至關(guān)重要。

介紹

使用python做web開發(fā)面臨的一個(gè)最大的問(wèn)題就是性能，在解決C10K問(wèn)題上顯的有點(diǎn)吃力。有些異步框架Tornado、Twisted、Gevent 等就是為了解決性能問(wèn)題。這些框架在性能上有些提升，但是也出現(xiàn)了各種古怪的問(wèn)題難以解決。

在python3.6中，官方的異步協(xié)程庫(kù)asyncio正式成為標(biāo)準(zhǔn)。在保留便捷性的同時(shí)對(duì)性能有了很大的提升,已經(jīng)出現(xiàn)許多的異步框架使用asyncio。

使用較早的異步框架是aiohttp，它提供了server端和client端，對(duì)asyncio做了很好的封裝。但是開發(fā)方式和最流行的微框架flask不同，flask開發(fā)簡(jiǎn)單，輕量，高效。

微服務(wù)是最近最火開發(fā)模式，它解決了復(fù)雜性問(wèn)題，提高開發(fā)效率，便于部署等優(yōu)點(diǎn)。

正是結(jié)合這些優(yōu)點(diǎn), 以Sanic為基礎(chǔ)，集成多個(gè)流行的庫(kù)來(lái)搭建微服務(wù)。 Sanic框架是和Flask相似的異步協(xié)程框架，簡(jiǎn)單輕量，并且性能很高。

本項(xiàng)目就是以Sanic為基礎(chǔ)搭建的微服務(wù)框架。

特點(diǎn)

使用sanic異步框架，簡(jiǎn)單，輕量，高效。

使用uvloop為核心引擎，使sanic在很多情況下單機(jī)并發(fā)甚至不亞于Golang。

使用asyncpg為數(shù)據(jù)庫(kù)驅(qū)動(dòng)，進(jìn)行數(shù)據(jù)庫(kù)連接，執(zhí)行sql語(yǔ)句執(zhí)行。

使用aiohttp為Client，對(duì)其他微服務(wù)進(jìn)行訪問(wèn)。

使用peewee為ORM，但是只是用來(lái)做模型設(shè)計(jì)和migration。

使用opentracing為分布式追蹤系統(tǒng)。

使用unittest做單元測(cè)試，并且使用mock來(lái)避免訪問(wèn)其他微服務(wù)。

使用swagger做API標(biāo)準(zhǔn)，能自動(dòng)生成API文檔。

使用

項(xiàng)目地址： sanic-ms

Example

服務(wù)端

使用sanic異步框架，有較高的性能，但是使用不當(dāng)會(huì)造成blocking, 對(duì)于有IO請(qǐng)求的都要選用異步庫(kù)。添加庫(kù)要慎重。
sanic使用uvloop異步驅(qū)動(dòng)，uvloop基于libuv使用Cython編寫，性能比nodejs還要高。

功能說(shuō)明：

啟動(dòng)前

@app.listener("before_server_start")
async def before_srver_start(app, loop):
    queue = asyncio.Queue()
    app.queue = queue
    loop.create_task(consume(queue, app.config.ZIPKIN_SERVER))
    reporter = AioReporter(queue=queue)
    tracer = BasicTracer(recorder=reporter)
    tracer.register_required_propagators()
    opentracing.tracer = tracer
    app.db = await ConnectionPool(loop=loop).init(DB_CONFIG)

創(chuàng)建DB連接池

創(chuàng)建Client連接

創(chuàng)建queue, 消耗span，用于日志追蹤

創(chuàng)建opentracing.tracer進(jìn)行日志追蹤

中間件

@app.middleware("request")
async def cros(request):
    if request.method == "POST" or request.method == "PUT":
        request["data"] = request.json
    span = before_request(request)
    request["span"] = span


@app.middleware("response")
async def cors_res(request, response):
    span = request["span"] if "span" in request else None
    if response is None:
        return response
    result = {"code": 0}
    if not isinstance(response, HTTPResponse):
        if isinstance(response, tuple) and len(response) == 2:
            result.update({
                "data": response[0],
                "pagination": response[1]
            })
        else:
            result.update({"data": response})
        response = json(result)
        if span:
            span.set_tag("http.status_code", "200")
    if span:
        span.set_tag("component", request.app.name)
        span.finish()
    return response

創(chuàng)建span, 用于日志追蹤

對(duì)response進(jìn)行封裝，統(tǒng)一格式

異常處理

對(duì)拋出的異常進(jìn)行處理，返回統(tǒng)一格式

任務(wù)

創(chuàng)建task消費(fèi)queue中對(duì)span，用于日志追蹤

異步處理

由于使用的是異步框架，可以將一些IO請(qǐng)求并行處理

Example:

async def async_request(datas):
    # async handler request
    results = await asyncio.gather(*[data[2] for data in datas])
    for index, obj in enumerate(results):
        data = datas[index]
        data[0][data[1]] = results[index]

@user_bp.get("/")
@doc.summary("get user info")
@doc.description("get user info by id")
@doc.produces(Users)
async def get_users_list(request, id):
    async with request.app.db.acquire(request) as cur:
        record = await cur.fetch(
            """ SELECT * FROM users WHERE id = $1 """, id)
        datas = [
            [record, "city_id", get_city_by_id(request, record["city_id"])]
            [record, "role_id", get_role_by_id(request, record["role_id"])]
        ]
        await async_request(datas)
        return record

get_city_by_id, get_role_by_id是并行處理。

相關(guān)連接

sanic

模型設(shè)計(jì) & ORM

Peewee is a simple and small ORM. It has few (but expressive) concepts, making it easy to learn and intuitive to use。ORM使用peewee, 只是用來(lái)做模型設(shè)計(jì)和migration, 數(shù)據(jù)庫(kù)操作使用asyncpg。

Example:

# models.py

class Users(Model):
    id = PrimaryKeyField()
    create_time = DateTimeField(verbose_name="create time",
                                default=datetime.datetime.utcnow)
    name = CharField(max_length=128, verbose_name="user"s name")
    age = IntegerField(null=False, verbose_name="user"s age")
    sex = CharField(max_length=32, verbose_name="user"s sex")
    city_id = IntegerField(verbose_name="city for user", help_text=CityApi)
    role_id = IntegerField(verbose_name="role for user", help_text=RoleApi)

    class Meta:
        db_table = "users"


# migrations.py

from sanic_ms.migrations import MigrationModel, info, db

class UserMigration(MigrationModel):
    _model = Users

    # @info(version="v1")
    # def migrate_v1(self):
    #     migrate(self.add_column("sex"))

def migrations():
    try:
        um = UserMigration()
        with db.transaction():
            um.auto_migrate()
            print("Success Migration")
    except Exception as e:
        raise e

if __name__ == "__main__":
    migrations()

運(yùn)行命令 python migrations.py

migrate_v1函數(shù)添加字段sex, 在BaseModel中要先添加name字段

info裝飾器會(huì)創(chuàng)建表migrate_record來(lái)記錄migrate，version每個(gè)model中必須唯一，使用version來(lái)記錄是否執(zhí)行過(guò)，還可以記錄author，datetime

migrate函數(shù)必須以migrate_開頭

相關(guān)連接

peewee

數(shù)據(jù)庫(kù)操作

asyncpg is the fastest driver among common Python, NodeJS and Go implementations使用asyncpg為數(shù)據(jù)庫(kù)驅(qū)動(dòng), 對(duì)數(shù)據(jù)庫(kù)連接進(jìn)行封裝, 執(zhí)行數(shù)據(jù)庫(kù)操作。
不使用ORM做數(shù)據(jù)庫(kù)操作，一個(gè)原因是性能，ORM會(huì)有性能的損耗，并且無(wú)法使用asyncpg高性能庫(kù)。另一個(gè)是單個(gè)微服務(wù)是很簡(jiǎn)單的，表結(jié)構(gòu)不會(huì)很復(fù)雜，簡(jiǎn)單的SQL語(yǔ)句就可以處理來(lái)，沒必要引入ORM。使用peewee只是做模型設(shè)計(jì)

Example:

sql = "SELECT * FROM users WHERE name=$1"
name = "test"
async with request.app.db.acquire(request) as cur:
    data = await cur.fetchrow(sql, name)

async with request.app.db.transaction(request) as cur:
    data = await cur.fetchrow(sql, name)

acquire() 函數(shù)為非事務(wù), 對(duì)于只涉及到查詢的使用非事務(wù)，可以提高查詢效率

tansaction() 函數(shù)為事務(wù)操作，對(duì)于增刪改必須使用事務(wù)操作

傳入request參數(shù)是為了獲取到span，用于日志追蹤

TODO 數(shù)據(jù)庫(kù)讀寫分離

相關(guān)連接

asyncpg
benchmarks

客戶端

使用aiohttp中的client，對(duì)客戶端進(jìn)行了簡(jiǎn)單的封裝，用于微服務(wù)之間訪問(wèn)。Don’t create a session per request. Most likely you need a session per application which performs all requests altogether.
A session contains a connection pool inside, connection reusage and keep-alives (both are on by default) may speed up total performance.

Example:

@app.listener("before_server_start")
async def before_srver_start(app, loop):
    app.client =  Client(loop, url="http://host:port")

async def get_role_by_id(request, id):
    cli = request.app.client.cli(request)
    async with cli.get("/cities/{}".format(id)) as res:
        return await res.json()

@app.listener("before_server_stop")
async def before_server_stop(app, loop):
    app.client.close()

對(duì)于訪問(wèn)不同的微服務(wù)可以創(chuàng)建多個(gè)不同的client，這樣每個(gè)client都會(huì)keep-alives

相關(guān)連接

aiohttp

日志 & 分布式追蹤系統(tǒng)

使用官方logging, 配置文件為logging.yml, sanic版本要0.6.0及以上。JsonFormatter將日志轉(zhuǎn)成json格式，用于輸入到ES

Enter OpenTracing: by offering consistent, expressive, vendor-neutral APIs for popular platforms, OpenTracing makes it easy for developers to add (or switch) tracing implementations with an O(1) configuration change. OpenTracing also offers a lingua franca for OSS instrumentation and platform-specific tracing helper libraries. Please refer to the Semantic Specification.

裝飾器logger

@logger(type="method", category="test", detail="detail", description="des", tracing=True, level=logging.INFO)
async def get_city_by_id(request, id):
    cli = request.app.client.cli(request)

type: 日志類型，如 method, route

category: 日志類別，默認(rèn)為app的name

detail: 日志詳細(xì)信息

description: 日志描述，默認(rèn)為函數(shù)的注釋

tracing: 日志追蹤，默認(rèn)為True

level: 日志級(jí)別，默認(rèn)為INFO

分布式追蹤系統(tǒng)

OpenTracing是以Dapper，Zipkin等分布式追蹤系統(tǒng)為依據(jù), 建立了統(tǒng)一的標(biāo)準(zhǔn)。

Opentracing跟蹤每一個(gè)請(qǐng)求，記錄請(qǐng)求所經(jīng)過(guò)的每一個(gè)微服務(wù)，以鏈條的方式串聯(lián)起來(lái)，對(duì)分析微服務(wù)的性能瓶頸至關(guān)重要。

使用opentracing框架，但是在輸出時(shí)轉(zhuǎn)換成zipkin格式。因?yàn)榇蠖鄶?shù)分布式追蹤系統(tǒng)考慮到性能問(wèn)題，都是使用的thrift進(jìn)行通信的，本著簡(jiǎn)單，Restful風(fēng)格的精神，沒有使用RPC通信。以日志的方式輸出, 可以使用fluentd, logstash等日志收集再輸入到Zipkin。Zipkin是支持HTTP輸入的。

生成的span先無(wú)阻塞的放入queue中，在task中消費(fèi)隊(duì)列的span。后期可以添加上采樣頻率。

對(duì)于DB，Client都加上了tracing

相關(guān)連接

opentracing
zipkin
jaeger

API接口

api文檔使用swagger標(biāo)準(zhǔn)。

Example:

from sanic_ms import doc

@user_bp.post("/")
@doc.summary("create user")
@doc.description("create user info")
@doc.consumes(Users)
@doc.produces({"id": int})
async def create_user(request):
    data = request["data"]
    async with request.app.db.transaction(request) as cur:
        record = await cur.fetchrow(
            """ INSERT INTO users(name, age, city_id, role_id)
                VALUES($1, $2, $3, $4, $5)
                RETURNING id
            """, data["name"], data["age"], data["city_id"], data["role_id"]
        )
        return {"id": record["id"]}

summary: api概要

description: 詳細(xì)描述

consumes: request的body數(shù)據(jù)

produces: response的返回?cái)?shù)據(jù)

tag: API標(biāo)簽

在consumes和produces中傳入的參數(shù)可以是peewee的model,會(huì)解析model生成API數(shù)據(jù), 在field字段的help_text參數(shù)來(lái)表示引用對(duì)象

http://host:ip/openapi/spec.json 獲取生成的json數(shù)據(jù)

相關(guān)連接

swagger

Response 數(shù)據(jù)

在返回時(shí)，不要返回sanic的response，直接返回原始數(shù)據(jù)，會(huì)在Middleware中對(duì)返回的數(shù)據(jù)進(jìn)行處理，返回統(tǒng)一的格式，具體的格式可以[查看]

單元測(cè)試

單元測(cè)試使用unittest。 mock是自己創(chuàng)建了MockClient，因?yàn)閡nittest還沒有asyncio的mock，并且sanic的測(cè)試接口也是發(fā)送request請(qǐng)求，所以比較麻煩. 后期可以使用pytest。

Example:

from sanic_ms.tests import APITestCase
from server import app

class TestCase(APITestCase):
    _app = app
    _blueprint = "visit"

    def setUp(self):
        super(TestCase, self).setUp()
        self._mock.get("/cities/1",
                       payload={"id": 1, "name": "shanghai"})
        self._mock.get("/roles/1",
                       payload={"id": 1, "name": "shanghai"})

    def test_create_user(self):
        data = {
            "name": "test",
            "age": 2,
            "city_id": 1,
            "role_id": 1,
        }
        res = self.client.create_user(data=data)
        body = ujson.loads(res.text)
        self.assertEqual(res.status, 200)

其中_blueprint為blueprint名稱

在setUp函數(shù)中，使用_mock來(lái)注冊(cè)mock信息, 這樣就不會(huì)訪問(wèn)真實(shí)的服務(wù)器, payload為返回的body信息

使用client變量調(diào)用各個(gè)函數(shù), data為body信息，params為路徑的參數(shù)信息，其他參數(shù)是route的參數(shù)

代碼覆蓋

coverage erase
coverage run --source . -m sanic_ms tests
coverage xml -o reports/coverage.xml
coverage2clover -i reports/coverage.xml -o reports/clover.xml
coverage html -d reports

coverage2colver 是將coverage.xml 轉(zhuǎn)換成 clover.xml，bamboo需要的格式是clover的。

相關(guān)連接

unittest
coverage

異常處理

使用 app.error_handler = CustomHander() 對(duì)拋出的異常進(jìn)行處理

Example:

from sanic_ms.exception import ServerError

@visit_bp.delete("/users/")
async def del_user(request, id):
    raise ServerError(error="內(nèi)部錯(cuò)誤",code=10500, message="msg")

code: 錯(cuò)誤碼，無(wú)異常時(shí)為0，其余值都為異常

message: 狀態(tài)碼信息

error: 自定義錯(cuò)誤信息

status_code: http狀態(tài)碼，使用標(biāo)準(zhǔn)的http狀態(tài)碼