Python が asyncio パッケージを使用して同時実行を処理する方法の詳細な説明

この記事は主に、同時実行性を処理するための Python の asyncio パッケージの使用に関する関連情報を詳しく紹介します。興味のある方は参考にしてください。

I/O と GIL のブロック

CPython の説明。はスレッドセーフではないため、一度に 1 つのスレッドのみが Python バイトコードを実行できるようにする Global Interpreter Lock (GIL) があります。したがって、Python プロセスは通常、複数の CPU コアを同時に使用できません。

ただし、ブロッキング I/O 操作を実行する標準ライブラリ内のすべての関数は、オペレーティング システムが結果を返すのを待っている間に GIL を解放します。これは、Python 言語レベルでマルチスレッドが可能であり、I/O 集中型の Python プログラムがこの利点を活用できることを意味します。1 つの Python スレッドがネットワーク応答を待機している間、ブロッキング I/O 関数が GIL を解放し、別のスレッドを実行します。 。

asyncio

このパッケージは、イベント ループ駆動のコルーチンを使用して同時実行性を実装します。 asyncio は式からのyieldを多用するため、古いバージョンのPythonとは互換性がありません。

asyncio パッケージで使用される「コルーチン」は、より厳密な定義です。 asyncio API に適したコルーチンは、定義本体で yield from を使用する必要がありますが、yield は使用できません。さらに、asyncio に適したコルーチンは呼び出し元によって駆動され、yield from を介して呼び出し元によって呼び出される必要があります。

asyncio.sleep(3) は 3 秒で完了するコルーチンを作成します。

loop.run_until_complete(future)、future が完了するまで実行します。パラメーターがコルーチン オブジェクトの場合は、ensure_future() 関数ラッパーを使用する必要があります。 loop.close() はイベントループを閉じます

例 2

import threading
import asyncio

@asyncio.coroutine
def hello():
  print('Start Hello', threading.currentThread())
  yield from asyncio.sleep(5)
  print('End Hello', threading.currentThread())

@asyncio.coroutine
def world():
  print('Start World', threading.currentThread())
  yield from asyncio.sleep(3)
  print('End World', threading.currentThread())

# 获取EventLoop:
loop = asyncio.get_event_loop()
tasks = [hello(), world()]
# 执行coroutine
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

説明:

1. asyncio.ensure_future(coro_or_future, *,loop=None): の実行をスケジュールする計画コルーチン オブジェクト、asyncio.Task オブジェクトを返します。 2.worker_fu.cancel(): コルーチンの実行をキャンセルし、CancelledError 例外をスローします。

3. asyncio.wait(): コルーチンのパラメーターは、future またはコルーチンで構成される反復可能なオブジェクトです。wait は各コルーチンを Task オブジェクトにラップします。

asyncio.Taskオブジェクトとthreading.Threadオブジェクトの比較

asyncio.Taskオブジェクトはthreading.Threadオブジェクトとほぼ同等です。

Task オブジェクトはコルーチンの駆動に使用され、Thread オブジェクトは呼び出し可能なオブジェクトの呼び出しに使用されます。
Task オブジェクトは自分でインスタンス化されるのではなく、コルーチンを asyncio.ensure_future(…) 関数またはloop.create_task(…) メソッドに渡すことによって取得されます。
取得した Task オブジェクトは実行するようにスケジュールされています。Thread インスタンスは start メソッドを呼び出して明示的に実行を指示する必要があります。タスクを終了したい場合は、Task.cancel() インスタンス メソッドを使用して、コルーチン内で CancelledError 例外をスローします。

スレッドとコルーチンの安全性の比較

スレッドを使用して重要なプログラミングを行った場合は、スケジューラーがいつでもスレッドを中断できるためです。プログラムの重要な部分を保護し、実行中に複数ステップの操作が中断されるのを防ぎ、データが無効な状態になるのを防ぐために、必ずロックを保持する必要があります。

コルーチンは、中断を防ぐためにデフォルトで完全に保護されます。プログラムの残りの部分を実行するには、明示的に出力する必要があります。コルーチンの場合、常に 1 つのコルーチンのみが実行されるため、ロックを保持したり、複数のスレッド間で操作を同期したりする必要はありません。制御を引き継ぎたい場合は、yield または yield from を使用して制御をスケジューラに戻すことができます。これが、コルーチンを安全にキャンセルできる理由です。定義上、コルーチンは一時停止されたイールドでのみキャンセルできるため、CancelledError 例外を処理し、クリーンアップ操作を実行できます。

Future (future)

通常、Future は自分で作成すべきではありませんが、同時実行フレームワーク (concurrent.futures または asyncio) によってのみインスタンス化できます。理由は簡単です。先物は最終的に何かが起こることを表しており、何かが起こることを確実に知る唯一の方法は、その実行がスケジュールされているかどうかだけです。

asyncio.Future

asyncio パッケージでは、BaseEventLoop.create_task(…) メソッドがコルーチンを受け取り、その実行時間をスケジュールし、asyncio.Future クラスのインスタンスでもある asyncio.Task インスタンスを返します。 Task は Future のサブクラスであり、コルーチンをラップするために使用されるためです。 asyncio.ensure_future(coro_or_future, *,loop=None)

この関数はコルーチンとフューチャーを統合します。最初のパラメーターは 2 つのいずれかになります。 Future オブジェクトまたは Task オブジェクトの場合は、変更せずに返されます。コルーチンの場合、async 関数はloop.create_task(...) メソッドを呼び出して Task オブジェクトを作成します。 loop= キーワード引数はオプションであり、イベント ループに渡すために使用されます。渡されない場合、async 関数は asyncio.get_event_loop() 関数を呼び出してループ オブジェクトを取得します。

BaseEventLoop.create_task(coro)このメソッドは、コルーチンの実行時間をスケジュールし、asyncio.Task オブジェクトを返します。

asyncio パッケージには、BaseEventLoop.run_until_complete(…) メソッドなど、asyncio.Task オブジェクトのパラメーターで指定されたコルーチンを自動的にラップする複数の関数があります。

asyncio.as_completed

为了集成进度条，我们可以使用的是 as_completed 生成器函数；幸好， asyncio 包提供了这个生成器函数的相应版本。

使用asyncio和aiohttp包

从 Python 3.4 起， asyncio 包只直接支持 TCP 和 UDP。如果想使用 HTTP 或其他协议，那么要借助第三方包 aiohttp 。

cc_list = ['China', 'USA']

@asyncio.coroutine
def get_flag(cc):
  url = '{}/{cc}/{cc}.gif'.format(BASE_URL, cc=cc.lower())
  resp = yield from aiohttp.request('GET', url)
  image = yield from resp.read()
  return image

@asyncio.coroutine
def download_one(name): 
  image = yield from get_flag(name) 
  save_flag(image, name.lower() + '.gif')
  return name

loop = asyncio.get_event_loop() 
wait_coro = asyncio.wait([download_one(cc) for cc in sorted(cc_list)]) 
res, _ = loop.run_until_complete(wait_coro) 
loop.close()

使用 asyncio 包时，我们编写的异步代码中包含由 asyncio 本身驱动的协程（即委派生成器），而生成器最终把职责委托给 asyncio 包或第三方库（如aiohttp）中的协程。这种处理方式相当于架起了管道，让 asyncio 事件循环（通过我们编写的协程）驱动执行低层异步 I/O 操作的库函数。

避免阻塞型调用

有两种方法能避免阻塞型调用中止整个应用程序的进程：
1. 在单独的线程中运行各个阻塞型操作
2. 把每个阻塞型操作转换成非阻塞的异步调用使用

多个线程是可以的，但是各个操作系统线程（Python 使用的是这种线程）消耗的内存达兆字节（具体的量取决于操作系统种类）。如果要处理几千个连接，而每个连接都使用一个线程的话，我们负担不起。

把生成器当作协程使用是异步编程的另一种方式。对事件循环来说，调用回调与在暂停的协程上调用 .send() 方法效果差不多。各个暂停的协程是要消耗内存，但是比线程消耗的内存数量级小。

上面的脚本为什么会很快

在上面的脚本中，调用 loop.run_until_complete 方法时，事件循环驱动各个download_one 协程，运行到第一个 yield from 表达式处时，那个表达式驱动各个get_flag 协程，然后在get_flag协程里面运行到第一个 yield from 表达式处时，调用 aiohttp.request(…)函数。这些调用都不会阻塞，因此在零点几秒内所有请求全部开始。

asyncio 的基础设施获得第一个响应后，事件循环把响应发给等待结果的 get_flag 协程。得到响应后， get_flag 向前执行到下一个 yield from 表达式处，调用resp.read() 方法，然后把控制权还给主循环。其他响应会陆续返回。所有 get_ flag 协程都获得结果后，委派生成器 download_one 恢复，保存图像文件。

async和await

为了简化并更好地标识异步IO，从Python 3.5开始引入了新的语法async和await，可以让coroutine的代码更简洁易读。

async和await是针对coroutine的新语法，要使用新的语法，只需要做两步简单的替换。
1. 把@asyncio.coroutine替换为async
2. 把yield from替换为await

例如：

@asyncio.coroutine
def hello():
  print("Hello world!")
  r = yield from asyncio.sleep(1)
  print("Hello again!")

等同于

async def hello():
  print("Hello world!")
  r = await asyncio.sleep(1)
  print("Hello again!")

网站请求实例

import asyncio
import aiohttp

urls = [
  'http://www.163.com/',
  'http://www.sina.com.cn/',
  'https://www.hupu.com/',
  'http://www.php.cn/'
]


async def get_url_data(u):
  """
  读取url的数据
  :param u:
  :return:
  """
  print('running ', u)
  async with aiohttp.ClientSession() as session:
    async with session.get(u) as resp:
      print(u, resp.status, type(resp.text()))
      # print(await resp.text())

  return resp.headers


async def request_url(u):
  """
  主调度函数
  :param u:
  :return:
  """
  res = await get_url_data(u)
  return res


loop = asyncio.get_event_loop()
task_lists = asyncio.wait([request_url(u) for u in urls])
all_res, _ = loop.run_until_complete(task_lists)
loop.close()

print(all_res)

以上がPython が asyncio パッケージを使用して同時実行を処理する方法の詳細な説明の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。