Python 語法校驗接口異步非阻塞實現(xiàn)

作者：鳥山明 2025-02-17 13:23:34

本文主要記錄線上案例中將同步阻塞代碼修改為異步非阻塞的過程，期間介紹 goInception 的基本使用、多進程實現(xiàn)踩的坑、測試 tornado 中異步的多種實現(xiàn)，最終使用協(xié)程，并對使用期間遇到的性能問題進行了簡單的分析。

引言

現(xiàn)象

背景：SQL 工單應用基于 tornado 框架實現(xiàn)，其中實現(xiàn)多個接口，包括語法校驗接口，其中語法校驗基于開源項目 goInception 實現(xiàn)。對于超長 SQL 或多實例的場景，經(jīng)常出現(xiàn)語法校驗超時的問題，原因是接口阻塞，IO 操作導致服務 block。

需求：改造接口實現(xiàn)，從同步阻塞修改為異步非阻塞，緩解語法校驗超時的問題。

當前實現(xiàn)

語法校驗接口

class StartSyncCheckHandler(tornado.web.RequestHandler):
  
  def post(self):
   ...
      return_data = mysql_check(job_option)
      self.finish(return_data)

接口中調(diào)用 goinception 實現(xiàn)語法校驗，goinception 使用的主要流程如下所示。

conn = self._get_inception_server_connection()
  cursor = conn.cursor()
  cursor.execute(self._get_inception_check_statement())
  query_result = cursor.fetchall()
  cursor.close()

由于 goinception 支持使用 MySQL 客戶端連接，因此和 MySQL 的使用方式相同，主要包括：

創(chuàng)建連接
創(chuàng)建 cursor
提交校驗
獲取校驗結果

使用 pymysql 創(chuàng)建連接，其中指定的 goinception 服務的 IP 和端口。

def _get_inception_server_connection():
  return pymysql.connect(
      host=GoInceptionServerConfig.mysql_ip,
      user=GoInceptionServerConfig.MySQL_User,
      passwd=GoInceptionServerConfig.MySQL_Password,
      port=GoInceptionServerConfig.MySQL_Port,
      charset=GoInceptionServerConfig.MySQL_Charset,
      db=GoInceptionServerConfig.Database_Name
  )

執(zhí)行校驗前生成提交給 goinception 的審核語句。

def _get_inception_check_statement(self):
        """
        獲取MySQL Inception使用的檢查語句
        :return:
        """
        backup_option = "--execute=0;--backup=0"
        run_option = "--check=1"
        inception_statement = """/*--user={inception_user};--password={inception_password};\
--host={inception_host};--port={inception_port};{run_option};{backup_option};*/ 
inception_magic_start;
{sql_script}
inception_magic_commit;
""".format(
            inception_user=self.mysql_user,
            inception_password=self.mysql_password,
            inception_host=self.mysql_ip,
            inception_port=self.mysql_port,
            run_option=run_option,
            backup_option=backup_option,
            sql_script=self.full_sql_script
        )
        return inception_statement

其中：

要求使用/* */將提交的信息括起來，其中每個參數(shù)通過分號分隔，包括要審核或執(zhí)行的語句，包括use database語句也要加分號，這一點與 MySQL 客戶端不同；
參數(shù)中的 IP 和端口是要校驗的 SQL 對應的數(shù)據(jù)庫；
指定--check=1;--execute=0，表示使用審核，不使用執(zhí)行；
goinception 支持語句塊的審核，要求通過執(zhí)行接口將要審核的 SQL 一次性提交，內(nèi)部拆分后一條條審核。其中inception_magic_start;作為語句塊的開始，inception_magic_start;作為語句塊的結束。

多進程啟動

tornado 默認使用單進程啟動，因此首先改用多進程啟動，具體實現(xiàn)是在啟動時指定進程數(shù)。

tornado.options.parse_command_line()
    app = Application()
    http_server = tornado.httpserver.HTTPServer(app)
    port = options.get("port", 8001)
    http_server.listen(port, "0.0.0.0")

    logging.warning("Server is running at http://0.0.0.0:%s" % port)

    tornado.ioloop.IOLoop.instance().start()
  
    # 多進程啟動
    # fork not available on windows。在windows上啟本動服務需注釋掉下面這行
    http_server.start(8)  # Forks multiple sub-processes

但是很快就發(fā)現(xiàn)了這種實現(xiàn)的缺點。主要包括：

并發(fā)數(shù)有上限，超過進程數(shù)后依然會發(fā)生阻塞等待，比如分庫分表語法校驗；
多個接口之間相互影響，當其他比較慢的接口用完了進程數(shù)，單實例的語法校驗也會發(fā)生阻塞等待。

下面是兩個案例的具體介紹。

案例1：并發(fā)數(shù)超過上限后語法校驗慢

時間：2023-09-05 10:28:37

現(xiàn)象：分庫分表語法校驗超時，16 個實例，每個實例 4 個 database，每個 database 256 個表，一共 16,384 個表。

日志

其中：

一批接收并處理 8 個請求，每個請求的執(zhí)行用時在 4s 左右；
每當一個請求返回后接收并處理下一個請求。

監(jiān)控顯示接口的 TP99 達到 9s，是接口實際執(zhí)行用時的兩倍左右。

監(jiān)控顯示 SQL 工單應用服務器 CPU 打滿，持續(xù)時間 30s 左右。

案例 2：其他接口慢導致執(zhí)行接口調(diào)用慢，如果調(diào)用語法校驗，同樣也會慢

時間：20230802 20:02

現(xiàn)象：執(zhí)行接口調(diào)用慢，判斷原因是空間檢測進程占用進程所致。

監(jiān)控顯示同一時間空間檢測接口的 TP99 超過 10s。

執(zhí)行接口正常情況下接口調(diào)用很快，主要是執(zhí)行一條 update SQL。查看執(zhí)行接口的日志，其中關鍵字 'xbp_id': 6044322 表示單號。

root@sql-inception-c3acb574:~# head 6044322.log 
2023-08-02 20:01:12,807 [MainThread:140461759665984] [sql] [start_execute_job:post:44] [INFO]- request_body_dict: {'job_uuid': '28182d96-905e-4807-aa4d-134f840cfe86', 'execute_method': 1, 'xbp_id': 6044322}
2023-08-02 20:01:12,807 [MainThread:140461759665984] [sql] [start_execute_job:post:50] [WARNING]- job_uuid=28182d96-905e-4807-aa4d-134f840cfe86, xbp_id=6044322
2023-08-02 20:01:12,888 [MainThread:140461759665984] [sql] [start_execute_job:post:66] [INFO]- return_data={'code': 0, 'message': 'start execute job success'}

工單 6044322 中有 37 個實例，調(diào)用執(zhí)行接口 37 次，全部日志顯示執(zhí)行接口調(diào)用的時間相差 20s，原因是接口阻塞。

root@sql-inception-c3acb574:~# cat /export/Logs/sql.log | grep "start_execute_job:post:44" | grep 6044322 | awk '{print $2}'
20:01:12,807
20:01:12,890
...
20:01:37,194
20:01:37,275
...
20:01:39,363
20:01:39,402

可以對比另一個工單 6051798 中有 64 個實例，調(diào)用執(zhí)行接口 64 次，全部日志顯示執(zhí)行接口調(diào)用的時間相差 0.2s。

root@sql-inception-c3acb574:~# cat /export/Logs/sql.log | grep "start_execute_job:post:44" | grep 6051798 | awk '{print $2}'
13:36:31,203
13:36:31,203
...
13:36:31,398
13:36:31,398

顯然，對于接口阻塞的問題，簡單實用多進程無法解決該問題，因此下面測試將接口改為異步非阻塞模式。

測試

準備

接口實現(xiàn)，其中調(diào)用 time.sleep(2) 模擬阻塞。

import datetime
import json
import time

import tornado.web


class AsyncHandler(tornado.web.RequestHandler):
    def post(self):
        data = json.loads(self.request.body)
        number = data.get("number")
        receive_time = datetime.datetime.now()
        print("=== receive number={} in {}".format(number, receive_time))
        print("==== {} enter ====".format(number))
        
        time.sleep(2)
        
        data = {
            "code": 0
        }
        print("==== {} exit ====".format(number))
        self.finish(data)

接口調(diào)用，其中使用線程池并發(fā)調(diào)用接口測試是否阻塞。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import datetime
import json

import requests

from common_helper.thread_pool import ThreadPool


def call_api(args):
    db_number = args[0] + 1
    print("send number={} in {}".format(db_number, datetime.datetime.now()))
    api = "async/"
    body = {
        "number": db_number
    }
    result = send_request(api, body=body)
    return result


def send_request(api, body):
    domain = "http://127.0.0.1:8000/"# 調(diào)用多接口時便于區(qū)分環(huán)境
    url = domain + api
    headers = {
        'content-type': "application/json",
    }
    result = requests.post(url=url, data=json.dumps(body), headers=headers)
    return result.content


def main():
    start_time = datetime.datetime.now()
    print(start_time)

    param_list = range(3)
    pool = ThreadPool(call_api, param_list)
    res_list = pool.run()
    print(res_list)
    end_time = datetime.datetime.now()
    print(end_time)
    use_time = end_time - start_time
    print("Use time={:.2f} seconds".format(use_time.total_seconds()))


if __name__ == '__main__':
    main()

阻塞

服務端日志

=== receive number=2 in2025-02-0920:03:15.001429
==== 2 enter ====
==== 2 exit ====
=== receive number=1in2025-02-0920:03:17.002924
==== 1 enter ====
==== 1 exit ====
=== receive number=3in2025-02-0920:03:19.008361
==== 3 enter ====
==== 3 exit ====

其中：

每次調(diào)用返回后執(zhí)行下一次調(diào)用，表明接口阻塞；
顯示 15、17、19 秒分別接收到請求，每個相差 2 秒。

客戶端日志

2025-02-0920:03:14.989916
send number=1in2025-02-0920:03:14.990300
send number=2in2025-02-0920:03:14.990630
send number=3in2025-02-0920:03:14.990811
[b'{"code": 0}', b'{"code": 0}', b'{"code": 0}']
2025-02-0920:03:21.012274
Use time=6.02 seconds

其中：

并發(fā)請求同時發(fā)送，時間都是 14 秒，這里可以留一個問題，客戶端發(fā)送請求但是服務端阻塞未處理時請求保存在哪里？
接口的 3 次調(diào)用總用時 6s，每次請求分別用時 2、4、6 秒。

接下來分別測試通過多種方法將阻塞接口修改為非阻塞。

async + await

將代碼中的阻塞方法 time.sleep 修改為非阻塞方法 tornado.gen.sleep。

在 Tornado 中，tornado.gen.sleep(2) 是一個協(xié)程，它會暫停當前協(xié)程的執(zhí)行，等待指定的時間（在本例中是 2 秒）后再恢復執(zhí)行。為了使用這個協(xié)程，我們需要在調(diào)用它的函數(shù)前面加上 async def 關鍵字，并在調(diào)用 tornado.gen.sleep(2) 時使用 await 關鍵字。

async def post(self):
      ...
        # time.sleep(2)
        await tornado.gen.sleep(2)
        ...
        # self.finish(data)  # Coroutine 'finish' is not awaited
        await self.finish(data)

服務端日志顯示接口處理過程中可以正常接收請求。

=== receive number=2 in2025-02-0920:28:59.343644
==== 2 enter ====
=== receive number=1in2025-02-0920:28:59.343680
==== 1 enter ====
=== receive number=3in2025-02-0920:28:59.343702
==== 3 enter ====
==== 2 exit ====
==== 1 exit ====
==== 3 exit ====

客戶端日志

2025-02-0920:28:59.332715
send number=1in2025-02-0920:28:59.333061
send number=2in2025-02-0920:28:59.333485
send number=3in2025-02-0920:28:59.333753
[b'{"code": 0}', b'{"code": 0}', b'{"code": 0}']
2025-02-0920:29:01.346989
Use time=2.01 seconds

測試顯示通過將阻塞方法 time.sleep 修改為非阻塞方法 tornado.gen.sleep 可以實現(xiàn)非阻塞。

@tornado.gen.coroutine

tornado 中可以使用 gen.coroutine 裝飾器實現(xiàn)異步，用于將生成器函數(shù)轉(zhuǎn)換成協(xié)程，其中使用 yield 關鍵字來暫停和恢復執(zhí)行。

import tornado.web
import tornado.gen


class AsyncHandler(tornado.web.RequestHandler):
    @tornado.gen.coroutine  # 異步、協(xié)程處理；增加并發(fā)量
    def post(self):
       ...
        # time.sleep(2)
        yield tornado.gen.sleep(2)
        ...
        self.finish(data)

其中：

用 tornado.gen.sleep 替換 time.sleep，fininsh 方法前不需要加 await，這種異步的實現(xiàn)可以像同步函數(shù)一樣編寫，便于理解與維護。

服務端日志

=== receive number=1 in 2025-02-09 20:42:11.404081
==== 1 enter ====
=== receive number=3 in 2025-02-09 20:42:11.404847
==== 3 enter ====
=== receive number=2 in 2025-02-09 20:42:11.404895
==== 2 enter ====
==== 1 exit ====
==== 3 exit ====
==== 2 exit ====

其中：

每次調(diào)用返回前開始執(zhí)行下一次調(diào)用，因此是非阻塞；

客戶端日志顯示同樣可以實現(xiàn)異步非阻塞。

2025-02-09 20:42:11.388831
send number=1 in 2025-02-09 20:42:11.389133
send number=2 in 2025-02-09 20:42:11.389564
send number=3 in 2025-02-09 20:42:11.389789
[b'{"code": 0}', b'{"code": 0}', b'{"code": 0}']
2025-02-09 20:42:13.407953
Use time=2.02 seconds

需要注意的是單純的 yield 并不能實現(xiàn)異步非阻塞，要求 yield 掛起的函數(shù)必須是非阻塞函數(shù)，比如這里如果還是使用 time.sleep 時依然阻塞。

@tornado.concurrent.run_on_executor

tornado 中的另一種實現(xiàn)是使用線程池后臺執(zhí)行，其中線程池使用單例模式。

from concurrent.futures import ThreadPoolExecutor

import tornado.web
import tornado.gen
import tornado.concurrent


class Executor(ThreadPoolExecutor):
    """
    單例模式實現(xiàn)線程池。大小為10
    """
    _instance = None

    def __new__(cls, *args, **kwargs):
        ifnot getattr(cls, "_instance", None):
            cls._instance = ThreadPoolExecutor(max_workers=10)
        return cls._instance


class AsyncHandler(tornado.web.RequestHandler):

    executor = Executor()

    @tornado.gen.coroutine  # 異步、協(xié)程處理；增加并發(fā)量
    def post(self):
       ...
        # time.sleep(2)
        yield self._process()
        ...
        self.finish(data)

    @tornado.concurrent.run_on_executor
    def _process(self):
        # RuntimeError: There is no current event loop in thread 'ThreadPoolExecutor-0_0'.
        # tornado.gen.sleep(2)
        time.sleep(2)

其中：

將 time.sleep 封裝在 _process 函數(shù)中， _process 函數(shù)調(diào)用前要加 yield，否則后臺執(zhí)行直接返回，不會等待結束；
不能使用 tornado.gen.sleep，否則報錯 RuntimeError: There is no current event loop in thread。

服務端日志

=== receive number=1 in2025-02-0920:49:27.726765
==== 1 enter ====
=== receive number=2in2025-02-0920:49:27.727211
==== 2 enter ====
=== receive number=3in2025-02-0920:49:27.727406
==== 3 enter ====
==== 2 exit ====
==== 3 exit ====
==== 1 exit ====

客戶端日志

2025-02-0920:49:27.716155
send number=1in2025-02-0920:49:27.716381
send number=2in2025-02-0920:49:27.716467
send number=3in2025-02-0920:49:27.716828
[b'{"code": 0}', b'{"code": 0}', b'{"code": 0}']
2025-02-0920:49:29.734317
Use time=2.02 seconds

測試顯示這種方式雖然可以實現(xiàn)異步非阻塞，但是本質(zhì)上還是線程池，因此無法滿足需求。

到這里可以再次明確需求，接口實現(xiàn)中有 IO 操作，需求是接口異步非阻塞，且需要等待 IO 返回結果，因此適合使用協(xié)程，原因是協(xié)程允許在執(zhí)行過程中暫停和恢復，從而實現(xiàn)異步編程。

而 tornado 中支持兩種異步的實現(xiàn)方式，包括 yield 掛起函數(shù)與類線程池，這里使用前者。

優(yōu)化

測試

class AsyncInceptionHandler(tornado.web.RequestHandler):
    def post(self):
        data = json.loads(self.request.body)
        number = data.get("number")
        receive_time = datetime.datetime.now()
        print("=== receive number={} in {}".format(number, receive_time))
        print("==== {} enter ====".format(number))

        inception_main()

        data = {
            "code": 0
        }
        print("==== {} exit ====".format(number))
        self.finish(data)
        
        
def inception_main():
    start_time = datetime.datetime.now()
    database_name = "cctest"
    sql_script = "select 1;"
  inception_test(database_name, sql_script)
    end_time = datetime.datetime.now()
    print(end_time.second - start_time.second)

    
def inception_test(database_name, sql_script):
    # 調(diào)用 goinception
    my_inception = GoInception(
        mysql_ip="x.x.x.x",
        mysql_port=3358,
        database_name=database_name,
        sql_script=sql_script,
    )

    check_data = my_inception.check_sql()
    return check_data


class GoInception(object):
    def check_sql(self):
          check_result = self._get_inception_check_result()
          returun check_result

    def _get_inception_check_result(self):
            """
            獲取MySQL Inception對腳本進行check操作后的結果
            :return:
            """
            cursor = self._get_inception_server_connection().cursor()
            cursor.execute(self._get_inception_check_statement())
            query_result = cursor.fetchall()
            return query_result

  @staticmethod
    def _get_inception_server_connection():
        """
        獲取MySQL Inception的連接
        :return:
        """
        return pymysql.connect(
            host=GoInceptionServerConfig.mysql_ip,
            user=GoInceptionServerConfig.mysql_user,
            passwd=GoInceptionServerConfig.mysql_password,
            port=GoInceptionServerConfig.mysql_port,
            charset=GoInceptionServerConfig.mysql_charset,
            db=GoInceptionServerConfig.database_name
        )

測試顯示接口單次調(diào)用執(zhí)行用時 5.4s

2025-02-16 21:46:56.962885
send number=1 in 2025-02-16 21:46:56.963171
[b'{"code": 0}']
2025-02-16 21:47:02.366053
Use time=5.40 seconds

服務端日志

Request received: /block_inception/
=== receive number=1 in 2025-02-16 21:46:56.972630
==== 1 enter ====
cursor = <pymysql.cursors.Cursor object at 0x7fad806061f0>
execute done
get result
error:
-54
==== 1 exit ====
Request processed: /block_inception/, Elapsed time: 5.39 seconds

線程池并發(fā) 3 次調(diào)用用時 6.25s，基本上等于單次調(diào)用用時的三倍。

2025-02-16 21:48:34.918864
send number=1 in 2025-02-16 21:48:34.919095
send number=2 in 2025-02-16 21:48:34.919488
send number=3 in 2025-02-16 21:48:34.919667
[b'{"code": 0}', b'{"code": 0}', b'{"code": 0}']
2025-02-16 21:48:52.591335
Use time=17.67 seconds

服務端日志顯示串行執(zhí)行，阻塞，單次執(zhí)行用時穩(wěn)定在 5s 左右。

Request received: /block_inception/
=== receive number=3in2025-02-1621:48:34.928843
==== 3 enter ====
cursor = <pymysql.cursors.Cursor object at 0x7f7de8836370>
execute done
get result
error:
8
==== 3 exit ====
Request processed: /block_inception/, Elapsed time: 7.51 seconds
Request received: /block_inception/
=== receive number=1in2025-02-1621:48:42.440652
==== 1 enter ====
cursor = <pymysql.cursors.Cursor object at 0x7f7de8816e20>
execute done
get result
error:
6
==== 1 exit ====
Request processed: /block_inception/, Elapsed time: 5.79 seconds
Request received: /block_inception/
=== receive number=2in2025-02-1621:48:48.230043
==== 2 enter ====
cursor = <pymysql.cursors.Cursor object at 0x7f7de8816d90>
execute done
get result
error:
4
==== 2 exit ====
Request processed: /block_inception/, Elapsed time: 4.36 seconds

下面將同步阻塞代碼修改為異步非阻塞模式。

aiomysql

使用 aiomyqsl + async + await

class AsyncInceptionHandler(tornado.web.RequestHandler):
    @tornado.gen.coroutine
    def post(self):
        data = json.loads(self.request.body)
        number = data.get("number")
        receive_time = datetime.datetime.now()
        print("=== receive number={} in {}".format(number, receive_time))
        print("==== {} enter ====".format(number))

        yield inception_main()

        data = {
            "code": 0
        }
        print("==== {} exit ====".format(number))
        self.finish(data)


asyncdef inception_main():
    start_time = datetime.datetime.now()
    database_name = "cctest"
    sql_script = "select 1;"
    await inception_test(database_name, sql_script)
    end_time = datetime.datetime.now()
    print(end_time.second - start_time.second)


asyncdef inception_test(database_name, sql_script):
    # 調(diào)用 goinception
    my_inception = GoInception(
        mysql_ip="x.x.x.x",
        mysql_port=3358,
        database_name=database_name,
        sql_script=sql_script,
    )

    check_data = await my_inception.check_sql()
    return check_data
    

class GoInception(object):
    asyncdef check_sql(self):
        check_result = await self._get_inception_check_result()
        return check_result
        
    asyncdef _get_inception_check_result(self):
        """
        獲取MySQL Inception對腳本進行check操作后的結果
        :return:
        """
        cursor = await self._get_inception_server_cursor()
        await cursor.execute(self._get_inception_check_statement())
        query_result = await cursor.fetchall()
        await cursor.close()
        return query_result
      
    @staticmethod
    asyncdef _get_inception_server_cursor():
        """
        獲取MySQL Inception的連接
        :return:
        """
        conn = await aiomysql.connect(
            host=GoInceptionServerConfig.mysql_ip,
            user=GoInceptionServerConfig.mysql_user,
            password=GoInceptionServerConfig.mysql_password,
            port=GoInceptionServerConfig.mysql_port,
            charset=GoInceptionServerConfig.mysql_charset,
            db=GoInceptionServerConfig.database_name
        )
        cursor = await conn.cursor()
        return cursor

測試結果

服務端日志顯示請求非阻塞，但是單次執(zhí)行用時穩(wěn)定在 9s 左右，顯示比阻塞請求的用時更長。

Request received: /inception/
=== receive number=1in2025-02-1621:50:16.826816
==== 1 enter ====
Request received: /inception/
=== receive number=3in2025-02-1621:50:16.828231
==== 3 enter ====
Request received: /inception/
=== receive number=2in2025-02-1621:50:16.828826
==== 2 enter ====
cursor = <aiomysql.cursors.Cursor object at 0x7f7de8836880>
cursor = <aiomysql.cursors.Cursor object at 0x7f7de8836640>
cursor = <aiomysql.cursors.Cursor object at 0x7f7de8836490>
execute done
get result
error:
8
==== 3 exit ====
Request processed: /inception/, Elapsed time: 7.76 seconds
execute done
get result
error:
10
==== 1 exit ====
Request processed: /inception/, Elapsed time: 9.38 seconds
execute done
get result
error:
11
==== 2 exit ====
Request processed: /inception/, Elapsed time: 10.68 seconds

客戶端日志

2025-02-1621:50:16.815495
send number=1in2025-02-1621:50:16.815727
send number=2in2025-02-1621:50:16.815810
send number=3in2025-02-1621:50:16.815895
[b'{"code": 0}', b'{"code": 0}', b'{"code": 0}']
2025-02-1621:50:27.509801
Use time=10.69 seconds

測試顯示優(yōu)化后不阻塞了，但是第一個請求需要等待最后一個請求結束才返回，整體執(zhí)行反倒變慢。

因此問題就是前面的請求為什么發(fā)生等待？

goinception

查看 goinception 的日志，其中顯示請求同時進入，但是也幾乎同時返回，因此懷疑慢不是應用的原因。

time="2025/02/16 21:50:17.334" level=info msg="con:79 new connection 100.124.212.72:48060" file=server.go func=onConn line=319
time="2025/02/16 21:50:17.340" level=info msg="con:80 new connection 100.124.212.72:48062" file=server.go func=onConn line=319
time="2025/02/16 21:50:17.341" level=info msg="con:81 new connection 100.124.212.72:48064" file=server.go func=onConn line=319
time="2025/02/16 21:50:24.848" level=info msg="con:80 close connection" file=server.go func=func1 line=321
time="2025/02/16 21:50:26.504" level=info msg="con:81 close connection" file=server.go func=func1 line=321
time="2025/02/16 21:50:27.753" level=info msg="con:79 close connection" file=server.go func=func1 line=321

使用 top 命令查看機器負載

其中：

MySQL 進程 CPU 使用率 92%；
goIncetpion 進程 CPU 使用率 54%。

上線后測試顯示機器負載不高的前提下可以實現(xiàn)異步非阻塞，但是 CPU 使用率高的問題還有待分析與解決。

下面使用 perf 分析 CPU。

首先使用 perf rerord 命令記錄程序運行期間的性能時間，默認將性能事件保存到 perf.data 文件中。

[root@test ~]# perf record -g -a sleep 15
[ perf record: Woken up 33 times to write data ]
[ perf record: Captured and wrote 8.452 MB perf.data (51875 samples) ]
[root@test ~]# 
[root@test ~]# ll -rth
-rw------- 1 root root 8.5M Feb 16 21:50 perf.data

然后使用 perf report 命令分析 perf record 保存到性能事件。

perf report -g

其中：

Symbol 表示函數(shù)名，其中 [.] 表示用戶空間函數(shù)，[k] 表示內(nèi)核空間；
Shared Object 表示函數(shù)所在的共享庫或所在的程序；
Command 表示進程名；
Children 表示該函數(shù)的 CPU 使用率；
Self 表示該函數(shù)的子函數(shù)的 CPU 使用率。

perf report --sort cpu -g --stdio

結合 top 與 perf 的結果，判斷接口返回慢的原因是機器負載高，機器負載高的原因主要包括 MySQL 連接處理與 goInception SQL 解析，具體有待進一步分析。

下面是用到的命令與參數(shù)。

其中：

perf record -g -a sleep 15

-g 表示保存函數(shù)調(diào)用的堆棧關系；
-a 表示記錄所有 CPU 上的數(shù)據(jù)；
sleep 15 表示 perf record 命令之后要運行的命令，sleep 15 命令會讓進程休眠 10 秒鐘，perf record 記錄執(zhí)行期間的所有事件。

perf report -g
-g 表示顯示調(diào)用堆棧，其中快捷鍵 E 展開，C 收起。
perf report --sort cpu -g --stdio
--sort cpu 表示按照 CPU 使用率排序，默認倒序；
--stdio 表示以文本模式顯示報告。

結論

SQL 工單應用中遇到語法校驗超時的問題，原因是接口同步阻塞，語法校驗最耗時的是 IO 操作，期間服務 block。

最開始的優(yōu)化方案是將應用啟動方式從單進程修改為多進程，但事實證明這種方式并不合理，原因是超過上限后依然阻塞，甚至多個接口之間相互影響。

因此將代碼從同步阻塞修改為異步非阻塞，tornado 中支持兩種異步的實現(xiàn)方式，包括 yield 掛起函數(shù)與類線程池，這里使用前者。當然要求 yield 掛起的函數(shù)支持非阻塞，這里使用 aiomysql 替換 pymysql、async def 替換 def。

修改后測試顯示接口不阻塞，但是第一個請求需要等待最后一個請求結束才返回，整體執(zhí)行反倒變慢。結合 top 與 perf 的結果，判斷接口返回慢的原因是機器負載高，機器負載高的原因主要包括 MySQL 連接處理與 goInception SQL 解析，具體有待進一步分析。

關于異步非阻塞與 CPU 使用率高的分析方法還需要深入學習，本文中暫未介紹。

責任編輯：華軒來源：丹柿小院