asyncio を利用して非同期http通信を実現する

背景

例えばエッジ端末で物体検知を行い続け、処理結果をサーバに送信したいとする.
detectionのループ処理中の間にHTTP POST処理を挟むと、レスポンスが帰ってくるまで処理が止まってしまう。
python標準のasync/awaitを利用することで、非同期通信が実現できる。その使い方をまとめる。

参考サイト

• http://iuk.hateblo.jp/entry/2017/01/27/173449
• https://qiita.com/meznat/items/c34fad95dab593f9bffa
• https://qiita.com/icoxfog417/items/07cbf5110ca82629aca0
• https://note.crohaco.net/2019/python-asyncio/
• https://docs.aiohttp.org/en/stable/client_quickstart.html#json-request
• https://docs.python.org/ja/3/library/asyncio-future.html#asyncio.Future.add_done_callback

環境

• python3.7.3

処理の流れ

import asyncio
import aiohttp

def say_something(future):    # 5
    print(future.result())
    return 'hello!'


async def nested(counter):
    return counter


async def test():
    await asyncio.sleep(2)    # 3
    return'aaaaaaaaaaaaaaaaaaaa'


async def main():
    rep = 0
    counter = 0
    while True:

        if (counter % 100 == 0) and (counter != 0):
            http_req = asyncio.create_task(test())    # 2
            http_req.add_done_callback(say_something)    # 4
            counter = 0
            print(f'rep: {rep}')
            rep += 1

        task = asyncio.create_task(nested(counter))

        print(await task)

        counter += 1


asyncio.run(main())    # 1

async とは

コルーチン(coroutin)を定義するための宣言。
コルーチンはサブルーチンのより一般的な形で、エントリポイントを複数個もつことができる。つまり、処理を中断してその場所から再開することができる。

イベントループとは

タスクスケジューラーのようなイメージで良い。イベントループに対してコルーチンを登録することで、実行・停止・再処理を自動で行ってくれる。

処理の中身

1. asyncio.runでイベントループにmainコルーチンを登録する。
2. mainコルーチンの内部では、特定条件の時にさらに別のコルーチンtest()をイベントループに登録する。test()は登録と同時に並列で処理が開始される。
3. test()内部では、await asyncio.sleep(2)によって「2秒間待つ」という処理が行われる。
4. main()の処理は止まらないが、今登録したtest()=http_reqが終了した際に実行するコールバック関数を定義する。
5. コールバック関数は、futureオブジェクト（http_reqの実体みたいなもの）の結果を標準出力に表示する。
   ※ asyncio.create_task()の返り値はfutureオブジェクトで、これは処理が完了した時にその返り値を自動的に自身に登録する。.result()で呼び出せる。
6. 特定条件下にない時も同様にイベントループに処理を登録して、実行する。で、whileループが継続する。

実装時の注意点

httpリクエストはrequestsライブラリを使わず、aiohttpというライブラリを使った。

$ pip install aiohttp

以下のような形でHTTPリクエストをコルーチンとして登録できる。

async def process_http_request(data):
    async with aiohttp.ClientSession() as session:
        async with session.post(URL, json=data) as response:
            return await response.text()