Python の Tornado フレームワークの AsyncHTTPClient を使用した非同期タスク-Python チュートリアル-php.cn

高性能サーバー Tornado
Python の Web フレームワークには多くの名前があり、それぞれに独自のメリットがあります。栄光がギリシャに属するように、偉大さはローマに属します。 Python の優雅さと WSGI の設計の組み合わせにより、Web フレームワークインターフェイスは何千年にもわたって統一されてきました。 WSGI はアプリケーションとサーバーを組み合わせます。 Django と Flask はどちらも gunicon と組み合わせて、アプリケーションを構築およびデプロイできます。

django や flask とは異なり、tornado は wsgi アプリケーションまたは wsgi サービスのいずれかになります。もちろん、tornado を選択する際の考慮事項は、そのシングルプロセス、シングルスレッドの非同期 IO ネットワークモードからもたらされます。高性能は魅力的ですが、トルネードは高性能だと謳っているのに、実際に使ってみるとなぜ実感できないのか、という疑問を友人からよく聞かれます。

実際、高いパフォーマンスは、Epoll (UNIX の場合は kqueue) に基づく Tornado の非同期ネットワーク IO から来ています。 Tornado のシングルスレッドメカニズムのため、サービスを誤ってブロック (ブロック) するコードを作成することが簡単です。パフォーマンスが向上しないだけでなく、実際にはパフォーマンスの急激な低下を引き起こします。したがって、トルネードの非同期的な使用を検討する必要があります。

Tornado の非同期使用法
つまり、Tornado の非同期には、非同期サーバーと非同期クライアントという 2 つの側面が含まれています。サーバーまたはクライアントに関係なく、特定の非同期モデルはコールバックとコルーチンに分割できます。特定のアプリケーションシナリオには明確な境界はありません。多くの場合、リクエストサービスには、他のサービスに対するクライアントの非同期リクエストも含まれます。

サーバー側の非同期メソッド
サーバー側の非同期は、トルネードリクエスト内で実行する必要がある時間のかかるタスクとして理解できます。ビジネスロジックに直接記述すると、サービス全体がブロックされる可能性があります。したがって、このタスクは非同期で処理できます。非同期処理を実現するには、yield 一時停止機能を使用する方法と、スレッドプールのような方法を使用する方法があります。同期の例をご覧ください:

リーリー

腹筋でテストしてください:

リーリー

qps はわずか 0.99 なので、1 秒あたり 1 つのリクエストを処理するものとして扱いましょう。

以下は非同期メソッドです:

リーリー

非同期タスクの実行時にタイムアウト 1 秒が選択されていますが、メインスレッドの復帰は依然として非常に高速です。腹圧テストは次のとおりです:

リーリー

上記の使用方法は、tornado の IO ループを介して、時間のかかるタスクをバックグラウンドで非同期計算に置くことができ、リクエストは他の計算を続行できます。ただし、時間のかかるタスクを完了した後に計算結果が必要になる場合もよくあります。この方法は現時点では機能しません。道路の前に道路があるはずです。非同期モードに切り替えるだけで済みます。以下はコルーチンを使用して書き換えられます:

リーリー

非同期処理が行われており、結果値も返されていることがわかります。

リーリー

QPS の改善は依然として明らかです。場合によっては、このコルーチン処理が同期よりも高速でない場合があります。同時実行の量が少ない場合、IO 自体によって生じるギャップは大きくありません。コルーチンと同期のパフォーマンスも同様です。例えば、ボルトと一緒に100メートル走ったら間違いなく負けますが、2メートル走ったらどちらが勝つかはまだ決まっていません。

Yield は関数コルーチンを一時停止しますが、ブロックのメインスレッドはありませんが、戻り値を処理する必要があるため、単一のリクエストと比較して応答が実行されるまで待つ時間があります。非同期およびコルーチンを使用するもう 1 つの方法は、メインスレッドの外側でスレッドプールを使用することです。スレッドプールは、future に依存します。 Python2 には追加のインストールが必要です。

以下のスレッドプールの利用方法を非同期処理に変更します。

リーリー

腹筋テストを再度実行します:

リーリー

qpsは即座に1912.78に達しました。同時に、サーバーのログが引き続き ping 結果を出力していることがわかります。

値を返すのも簡単です。次に、使用インターフェースを切り替えます。 tornado の gen モジュールで with_timeout 関数を使用します (この関数は tornado>3.2 バージョンである必要があります)。

リーリー

スレッドプール方式では、トルネードのyieldを利用して関数を一時停止し、コルーチン処理を実装することもできます。メインスレッドをブロックすることなく、時間のかかるタスクの結果を取得できます。

リーリー

qps は 116 です。yield コルーチン手法を使用すると、非応答の場合の約 10 分の 1 にすぎません。主な理由は、コルーチンが結果を返す前にタスクの完了を待つ必要があることです。

前者の方法は、網を投げてから作業を終了するので、当然時間がかかります。ネットが閉まるのを待ちます。もちろん、それでも同期方式よりも何百倍も速いのですが、結局のところ、網を投げるほうが、一匹ずつ漁をするよりも速いのです。

使用する具体的なメソッドはビジネスによって異なります。多くの場合、コールバックをネストする必要がある場合は、コールバックを処理する必要があります。最適化するのはビジネスロジックまたは製品ロジックである必要があります。 yield メソッドは非常にエレガントで、書き込みメソッドは非同期でも論理同期でも書けるという優れものですが、当然ながらある程度のパフォーマンスも低下します。

异步多样化
Tornado异步服务的处理大抵如此。现在异步处理的框架和库也很多，借助redis或者celery等，也可以把tonrado中一些业务异步化，放到后台执行。

此外，Tornado还有客户端异步功能。该特性主要是在于 AsyncHTTPClient的使用。此时的应用场景往往是tornado服务内，需要针对另外的IO进行请求和处理。顺便提及，上述的例子中，调用ping其实也算是一种服务内的IO处理。接下来，将会探索一下AsyncHTTPClient的使用，尤其是使用AsyncHTTPClient上传文件与转发请求。

异步客户端
前面了解Tornado的异步任务的常用做法，姑且归结为异步服务。通常在我们的服务内，还需要异步的请求第三方服务。针对HTTP请求，Python的库Requests是最好用的库，没有之一。官网宣称：HTTP for Human。然而，在tornado中直接使用requests将会是一场恶梦。requests的请求会block整个服务进程。

上帝关上门的时候，往往回打开一扇窗。Tornado提供了一个基于框架本身的异步HTTP客户端（当然也有同步的客户端）--- AsyncHTTPClient。

AsyncHTTPClient 基本用法
AsyncHTTPClient是 tornado.httpclinet 提供的一个异步http客户端。使用也比较简单。与服务进程一样，AsyncHTTPClient也可以callback和yield两种使用方式。前者不会返回结果，后者则会返回response。

如果请求第三方服务是同步方式，同样会杀死性能。

class SyncHandler(tornado.web.RequestHandler):
  def get(self, *args, **kwargs):

    url = 'https://api.github.com/'
    resp = requests.get(url)
    print resp.status_code

    self.finish('It works')

ログイン後にコピー

使用ab测试大概如下：

Document Path:     /sync
Document Length:    5 bytes

Concurrency Level:   5
Time taken for tests:  10.255 seconds
Complete requests:   5
Failed requests:    0
Total transferred:   985 bytes
HTML transferred:    25 bytes
Requests per second:  0.49 [#/sec] (mean)
Time per request:    10255.051 [ms] (mean)
Time per request:    2051.010 [ms] (mean, across all concurrent requests)
Transfer rate:     0.09 [Kbytes/sec] received

ログイン後にコピー

性能相当慢了，换成AsyncHTTPClient再测：

class AsyncHandler(tornado.web.RequestHandler):
  @tornado.web.asynchronous
  def get(self, *args, **kwargs):

    url = 'https://api.github.com/'
    http_client = tornado.httpclient.AsyncHTTPClient()
    http_client.fetch(url, self.on_response)
    self.finish('It works')

  @tornado.gen.coroutine
  def on_response(self, response):
    print response.code

ログイン後にコピー

qps 提高了很多

Document Path:     /async
Document Length:    5 bytes

Concurrency Level:   5
Time taken for tests:  0.162 seconds
Complete requests:   5
Failed requests:    0
Total transferred:   985 bytes
HTML transferred:    25 bytes
Requests per second:  30.92 [#/sec] (mean)
Time per request:    161.714 [ms] (mean)
Time per request:    32.343 [ms] (mean, across all concurrent requests)
Transfer rate:     5.95 [Kbytes/sec] received

ログイン後にコピー

同样，为了获取response的结果，只需要yield函数。

class AsyncResponseHandler(tornado.web.RequestHandler):
  @tornado.web.asynchronous
  @tornado.gen.coroutine
  def get(self, *args, **kwargs):

    url = 'https://api.github.com/'
    http_client = tornado.httpclient.AsyncHTTPClient()
    response = yield tornado.gen.Task(http_client.fetch, url)
    print response.code
    print response.body

ログイン後にコピー

AsyncHTTPClient 转发
使用Tornado经常需要做一些转发服务，需要借助AsyncHTTPClient。既然是转发，就不可能只有get方法，post，put，delete等方法也会有。此时涉及到一些 headers和body，甚至还有https的waring。

下面请看一个post的例子， yield结果，通常，使用yield的时候，handler是需要 tornado.gen.coroutine。

headers = self.request.headers
body = json.dumps({'name': 'rsj217'})
http_client = tornado.httpclient.AsyncHTTPClient()

resp = yield tornado.gen.Task(
  self.http_client.fetch, 
  url,
  method="POST", 
  headers=headers,
  body=body, 
  validate_cert=False)

ログイン後にコピー

AsyncHTTPClient 构造请求
如果业务处理并不是在handlers写的，而是在别的地方，当无法直接使用tornado.gen.coroutine的时候，可以构造请求，使用callback的方式。

body = urllib.urlencode(params)
req = tornado.httpclient.HTTPRequest(
 url=url, 
 method='POST', 
 body=body, 
 validate_cert=False) 

http_client.fetch(req, self.handler_response)

def handler_response(self, response):

  print response.code

ログイン後にコピー

用法也比较简单，AsyncHTTPClient中的fetch方法，第一个参数其实是一个HTTPRequest实例对象，因此对于一些和http请求有关的参数，例如method和body，可以使用HTTPRequest先构造一个请求，再扔给fetch方法。通常在转发服务的时候，如果开起了validate_cert，有可能会返回599timeout之类，这是一个warning，官方却认为是合理的。

AsyncHTTPClient 上传图片
AsyncHTTPClient 更高级的用法就是上传图片。例如服务有一个功能就是请求第三方服务的图片OCR服务。需要把用户上传的图片，再转发给第三方服务。

@router.Route('/api/v2/account/upload')
class ApiAccountUploadHandler(helper.BaseHandler):
  @tornado.gen.coroutine
  @helper.token_require
  def post(self, *args, **kwargs):
    upload_type = self.get_argument('type', None)

    files_body = self.request.files['file']

    new_file = 'upload/new_pic.jpg'
    new_file_name = 'new_pic.jpg'

    # 写入文件
    with open(new_file, 'w') as w:
      w.write(file_['body'])

    logging.info('user {} upload {}'.format(user_id, new_file_name))

    # 异步请求 上传图片
    with open(new_file, 'rb') as f:
      files = [('image', new_file_name, f.read())]

    fields = (('api_key', KEY), ('api_secret', SECRET))

    content_type, body = encode_multipart_formdata(fields, files)

    headers = {"Content-Type": content_type, 'content-length': str(len(body))}
    request = tornado.httpclient.HTTPRequest(config.OCR_HOST,
                         method="POST", headers=headers, body=body, validate_cert=False)

    response = yield tornado.httpclient.AsyncHTTPClient().fetch(request)

def encode_multipart_formdata(fields, files):
  """
  fields is a sequence of (name, value) elements for regular form fields.
  files is a sequence of (name, filename, value) elements for data to be
  uploaded as files.
  Return (content_type, body) ready for httplib.HTTP instance
  """
  boundary = '----------ThIs_Is_tHe_bouNdaRY_$'
  crlf = '\r\n'
  l = []
  for (key, value) in fields:
    l.append('--' + boundary)
    l.append('Content-Disposition: form-data; name="%s"' % key)
    l.append('')
    l.append(value)
  for (key, filename, value) in files:
    filename = filename.encode("utf8")
    l.append('--' + boundary)
    l.append(
        'Content-Disposition: form-data; name="%s"; filename="%s"' % (
          key, filename
        )
    )
    l.append('Content-Type: %s' % get_content_type(filename))
    l.append('')
    l.append(value)
  l.append('--' + boundary + '--')
  l.append('')
  body = crlf.join(l)
  content_type = 'multipart/form-data; boundary=%s' % boundary
  return content_type, body


def get_content_type(filename):
  import mimetypes

  return mimetypes.guess_type(filename)[0] or 'application/octet-stream'

ログイン後にコピー

对比上述的用法，上传图片仅仅是多了一个图片的编码。将图片的二进制数据按照multipart 方式编码。编码的同时，还需要把传递的相关的字段处理好。相比之下，使用requests 的方式则非常简单：

files = {}
f = open('/Users/ghost/Desktop/id.jpg')
files['image'] = f
data = dict(api_key='KEY', api_secret='SECRET')
resp = requests.post(url, data=data, files=files)
f.close()
print resp.status_Code

ログイン後にコピー

总结
通过AsyncHTTPClient的使用方式，可以轻松的实现handler对第三方服务的请求。结合前面关于tornado异步的使用方式。无非还是两个key。是否需要返回结果，来确定使用callback的方式还是yield的方式。当然，如果不同的函数都yield，yield也可以一直传递。这个特性，tornado的中的tornado.auth 里面对oauth的认证。

大致就是这样的用法。