【ステップアップ】「Pythonの実践」簡単速習‼【非同期処理 asynico/応用⑫】

こんにちはヤク学長です。
データサイエンティスト兼ファーマシストで、アルゴリズムやBI開発を行っています。

本記事の目的は、「pythonの基本操作を知る」ことを目的としています。

【ステップアップ】「Pythonの実践」簡単速習‼【キューイングシステム/応用⑪】

【本記事のもくじ】

まず、「Python」に真剣に取り組むための概要を解説します。
下記の方法で、簡単に概要を抑えることができます。

• １．非同期処理 asynico

それでは、上から順番に見ていきます。
なお、本上記の方法を順番に抑えれば成果が出ます。

記事の内容は「転載 & 引用OK」問題ありません。

１．非同期処理 asynico

asyncioは、Python 3.4以降で導入された非同期処理のための標準ライブラリです。asyncioを使用すると、単一のスレッド内で複数のタスクを実行できます。これにより、I/Oバウンドタスクのパフォーマンスが向上し、マルチスレッドプログラミングよりも簡単になります。

asyncioには、イベントループ（event loop）、コルーチン（coroutine）、タスク（task）などのコンセプトがあります。イベントループは、タスクを実行し、コルーチンは非同期関数のようなもので、タスクはコルーチンをラップするものです。

以下は、asyncioを使用して、3つの非同期タスクを実行する例です。

import asyncio

async def print_message(message):
print(message)
await asyncio.sleep(1)

async def main():
task1 = asyncio.create_task(print_message("Hello"))
task2 = asyncio.create_task(print_message("world"))
task3 = asyncio.create_task(print_message("from asyncio!"))

await task1
await task2
await task3

asyncio.run(main())

この例では、asyncio.create_task()関数を使用して、3つの非同期タスクを作成し、それらをawaitで待機しています。各タスクは、1秒間スリープしてからメッセージを出力します。

このように、asyncioを使用すると、非同期的に複数のタスクを実行できます。ただし、タスクがI/Oバウンドでない場合や、CPUバウンドのタスクを非同期で実行する場合は、asyncioよりもスレッドやプロセスを使用した方が効率的な場合があります。

スレッドとマルチプロセスのI/OバウンドとCPUバウンド

スレッドとは、プロセス内で複数のスレッドを作成し、各スレッドが個別の処理を行うことで並列処理を実現するものです。スレッドを使用する場合、各スレッドの実行コンテキストを管理するためのオーバーヘッドがありますが、メモリ消費量が少ないため、I/Oバウンドのタスクに適しています。つまり、入出力待ちが発生するタスクに適しています。例えば、ネットワークリクエストやデータベースアクセスなどです。

一方、マルチプロセスは、複数のプロセスを使用して並列処理を実現するもので、各プロセスが個別のアドレス空間を持つため、メモリ消費量が多くなります。CPUバウンドのタスクに適しており、CPUを多用する計算に向いています。例えば、画像処理や数値計算などです。

Pythonにおいて、スレッドは threading モジュールを使用して、マルチプロセスは multiprocessing モジュールを使用して実装することができます。また、Python 3.5以降では、非同期処理に対応するための asyncio モジュールが提供されており、非同期I/O処理に適したスレッドとは異なる非同期処理を実現することができます。

非同期処理とは

非同期処理は、複数の処理を同時に実行できるプログラミングの手法です。一つの処理が完了するまで、別の処理が実行されることで、プログラム全体の性能を向上させることができます。例えば、ネットワーク通信を行う際には、データの送受信に時間がかかるため、その間に別の処理を実行できるようにすることで、より効率的にプログラムを実行できます。

非同期処理には、コールバック、Promise、async/await などの手法があります。これらの手法を使用することで、コードをより簡潔に書くことができます。また、Python には、asyncio モジュールが標準で用意されており、非同期処理を簡単に実装することができます。

ジェネレーターベースのコルーチン

ジェネレーターベースのコルーチンは、Python 2.5で導入された機能です。コルーチンは、ジェネレーターを使って非同期処理を実現するための方法です。ジェネレーターを使うことで、コードを複数の部分に分割し、コードが実行されるたびに、実行の状態を保持することができます。

ジェネレーターベースのコルーチンは、通常のジェネレーターと同様に、yield文を使用して値を生成しますが、ジェネレーターからの値の返し方が異なります。ジェネレーターベースのコルーチンでは、ジェネレーターがyield文で値を返すと、コードの呼び出し元に制御が戻るのではなく、ジェネレーター内で待機している別のジェネレーターに制御が移ります。これにより、非同期処理を実現することができます。

ジェネレーターベースのコルーチンは、Python 3.5以降のバージョンではasync/await構文に置き換えられ、asyncioモジュールでサポートされるようになりました。async/await構文を使うことで、より簡潔でわかりやすい非同期コードを書くことができます。

yield from文

yield from 文は、Python 3.3 から導入された機能で、ジェネレータを簡潔かつ効果的に書くための方法です。yield from を使用すると、ネストされたジェネレータの処理を簡単に実装できます。

yield from の基本的な構文は次のとおりです。

yield from ジェネレータ

この文は、渡されたジェネレータからの値を返すだけでなく、制御をそのジェネレータに移します。つまり、yield from が完了するまで、現在のジェネレータは中断され、渡されたジェネレータが実行されます。ジェネレータが終了すると、制御は元のジェネレータに戻り、yield from の次の行から実行が再開されます。

yield from は、通常の yield 文と比較して、いくつかの利点があります。最も重要な利点は、ネストされたコルーチンを処理する際に、入力と出力を処理するための冗長なコードを書かなくてもよいということです。また、yield from を使用することで、コルーチンのエラーハンドリングが簡単になります。

以下は、yield from を使用したジェネレータの例です。

def subgen():
while True:
received = yield
print(received)

def grouper():
while True:
group = yield from subgen()
print(group)

g = grouper()
next(g)
g.send("Hello world!")
g.send("Goodbye world!")

コルーチンとネイティブコルーチン

Pythonにおいて、コルーチンは非同期プログラミングを実現するための機能の1つであり、イテレーターと同様に値を生成することができますが、イテレーターと異なり、値の生成と値の受け取りを両方の方向に行うことができます。 つまり、コルーチンは関数として呼び出すことができ、処理を一時停止して値を送信したり、受信したりできることができます。

Python 3.5からは、async / awaitキーワードを使ったネイティブコルーチンが導入され、コルーチンが非同期処理を行う際に簡単に扱うことができるようになりました。ネイティブコルーチンは、async defキーワードで宣言され、awaitキーワードでコルーチン内で別のコルーチンを呼び出すことができます。ネイティブコルーチンを使用すると、非同期処理を簡単に扱うことができます。

ジェネレーターベースのコルーチンは、yieldキーワードを使用して値を生成し、send（）メソッドを使用して値を送信できます。ネイティブコルーチンでは、awaitキーワードを使用して別のコルーチンを呼び出し、値を返すことができます。ジェネレーターベースのコルーチンは、Python 3.3以降で利用可能ですが、ネイティブコルーチンを使用する方が一般的です。

asynicoを使用する場面はどんな場合？

asyncioは、主にI/Oバウンドなタスクを実行するために使用されます。つまり、ネットワークリクエスト、ファイル操作、データベースアクセスなど、システムリソースに依存する処理を非同期に処理することができます。

また、CPUバウンドなタスク（CPUを多く消費するタスク）もasyncioで実行できますが、他のマルチスレッド/マルチプロセスライブラリ（例えば、concurrent.futures）を使用する場合に比べて、asyncioのパフォーマンスが低下する可能性があることに注意してください。

asynico.Lock

asyncio.Lockは、asyncioモジュールに含まれるクラスで、複数のコルーチンが同時にアクセスできないようにするために使用されます。これは、多くのコルーチンが同時に同じリソースを使用する場合、データ競合が発生することがあります。asyncio.Lockは、コルーチンがリソースにアクセスする前にロックを取得し、リソースが解放されるまで他のコルーチンが待機するようにします。

以下は、asyncio.Lockを使用して、2つのコルーチンが同じリソースにアクセスできないようにする例です。

import asyncio

async def coro_1(lock):
async with lock:
print("coro_1 has acquired the lock")
await asyncio.sleep(2)
print("coro_1 has released the lock")

async def coro_2(lock):
async with lock:
print("coro_2 has acquired the lock")
await asyncio.sleep(2)
print("coro_2 has released the lock")

async def main():
lock = asyncio.Lock()
await asyncio.gather(coro_1(lock), coro_2(lock))

asyncio.run(main())

asynico.Event

asyncio.Eventは、Pythonの非同期ライブラリであるasyncioで提供されているイベントオブジェクトです。イベントオブジェクトは、一つのスレッドまたはタスクからセットされ、もう一つのスレッドまたはタスクから待機されるオブジェクトで、多くの場合、同期化のために使用されます。

asyncio.Eventは、以下のようなメソッドを提供します。

• set()：イベントの状態をTrueにセットします。
• clear()：イベントの状態をFalseにセットします。
• wait()：イベントの状態がTrueになるまでブロックし、その後、イベントの状態をFalseにリセットします。
• is_set()：イベントの状態がTrueである場合はTrueを返します。

以下は、asyncio.Eventを使用した簡単な例です。

import asyncio

async def coro1(event):
print('coro1 waiting')
await event.wait()
print('coro1 done')

async def coro2(event):
print('coro2 waiting')
await event.wait()
print('coro2 done')

async def main():
event = asyncio.Event()
await asyncio.gather(coro1(event), coro2(event))
event.set()

asyncio.run(main())

asynico.condition

asyncio.Conditionは、複数のタスクが同期的に協力して共有状態を更新する必要がある場合に便利なツールです。このオブジェクトは、複数のタスクが相互排除制御を使用して、条件を満たすまで待機できるようにします。

asyncio.Conditionは、async withステートメントを使用して簡単に作成できます。以下は、条件変数の使用例です。

import asyncio

async def consumer(condition):
async with condition:
await condition.wait()
print('Consumer consumed the resource')

async def producer(condition):
await asyncio.sleep(1)
async with condition:
print('Producer produced the resource')
condition.notify_all()

async def main():
condition = asyncio.Condition()
consumer_task = asyncio.create_task(consumer(condition))
producer_task = asyncio.create_task(producer(condition))
await asyncio.gather(consumer_task, producer_task)

asyncio.run(main())

asinico.Semapphore

asyncio.Semaphore は、同時にアクセスできる数が制限されたリソースのアクセスを制御するための同期プリミティブです。セマフォは、資源の数を指定された値に設定することで作成されます。acquire() メソッドを呼び出すことで、リソースを1つ消費します。もしセマフォにリソースがない場合、acquire() はリソースが利用可能になるまでブロックされます。release() メソッドを呼び出すことで、リソースを1つ解放し、他のタスクがそのリソースにアクセスできるようになります。

以下は、asyncio.Semaphore を使用した簡単な例です。

import asyncio

async def worker(semaphore, task_name):
# セマフォを取得する
async with semaphore:
print(f"{task_name} started.")
await asyncio.sleep(1)
print(f"{task_name} finished.")

async def main():
# セマフォを初期化する
semaphore = asyncio.Semaphore(2)

# 複数のタスクを実行する
tasks = [
asyncio.create_task(worker(semaphore, "Task 1")),
asyncio.create_task(worker(semaphore, "Task 2")),
asyncio.create_task(worker(semaphore, "Task 3")),
asyncio.create_task(worker(semaphore, "Task 4"))
]
await asyncio.gather(*tasks)

asyncio.run(main())

asinico.Queue

asyncio.Queueは、Pythonの標準ライブラリで提供される非同期キューの実装です。このキューは、複数のタスクからアクセスされる可能性があるデータの共有に使用されます。キューは先入れ先出し（FIFO）で、内部的にはリストを使用して実装されています。

例えば、非同期的にデータを収集し、別のタスクでデータを処理する場合に、asyncio.Queueを使用できます。データを収集するタスクは、収集したデータをキューに追加し、データを処理するタスクは、キューからデータを取得して処理します。asyncio.Queueは、タスク間でのデータの共有と同期に役立ちます。

以下は、asyncio.Queueを使用して、非同期的にキューにデータを追加し、別のタスクでデータを取得する簡単な例です。

import asyncio

async def producer(queue):
for i in range(10):
await asyncio.sleep(1) # simulate I/O bound work
await queue.put(i)
await queue.put(None) # indicate end of stream

async def consumer(queue):
while True:
item = await queue.get()
if item is None:
break
print(item)
queue.task_done()

async def main():
queue = asyncio.Queue()
producer_task = asyncio.create_task(producer(queue))
consumer_task = asyncio.create_task(consumer(queue))
await queue.join()
producer_task.cancel()
await producer_task
await consumer_task

asyncio.run(main())

asinico.Future

asyncio.Future は、concurrent.futures.Future の非同期版であり、Python 3.4 以降で利用可能です。非同期処理の中で、将来的に完了する結果を表現するために使用されます。通常、asyncio.create_task() などで非同期関数をタスクとして実行した場合、実行されたタスクが完了すると、その結果は自動的に非同期 Future オブジェクトにセットされます。非同期 Future オブジェクトには await を使用してアクセスすることができ、結果が利用可能になるまで非同期にブロックすることができます。

例えば、以下のように、asyncio.sleep() 関数を使用して、1秒後に整数値を返す非同期関数を定義し、その関数をタスクとして実行すると、タスクが完了した後、非同期 Future オブジェクトの result() メソッドで結果を取得することができます。

import asyncio

async def async_func():
await asyncio.sleep(1)
return 123

async def main():
task = asyncio.create_task(async_func())
await task
print(task.result())

asyncio.run(main())

asinico.cell_soonとasinico.call_later

asyncio.call_soon と asyncio.call_later は、非同期関数を実行するための asyncio モジュールの関数です。 call_soon 関数は、非同期関数をすぐに実行し、call_later 関数は、非同期関数を指定した時間（秒数）後に実行します。

具体的には、以下のように使用します。

import asyncio

async def my_coroutine():
print('Coroutine is running')

def main():
loop = asyncio.get_event_loop()
loop.call_soon(my_coroutine())
loop.call_later(2, my_coroutine())
loop.run_forever()

if __name__ == '__main__':
main()

この例では、 my_coroutine 関数を2回実行します。1回目は call_soon 関数によってすぐに実行され、2回目は call_later 関数によって2秒後に実行されます。run_forever メソッドは、イベントループが終了するまで実行されます。

注意点としては、call_soon 関数と call_later 関数に渡す引数は、コルーチンを直接渡すのではなく、コルーチンオブジェクトを渡す必要がある点です。

asinicoのチェーン

asynicoのチェーンとは、非同期処理において、複数の処理を連鎖的に実行する仕組みです。asynicoでは、コルーチンを作成して、それらをチェーンとしてつなげることで、シンプルかつ柔軟な非同期処理を実現することができます。

例えば、以下のような2つのコルーチンがあるとします。

async def first_coroutine():
# do some processing
result = await some_async_function()
# do some more processing
return result

async def second_coroutine():
# do some processing
result = await first_coroutine()
# do some more processing
return result

上記の例では、first_coroutineというコルーチンが定義されており、その中で非同期関数を呼び出していることがわかります。また、second_coroutineというコルーチンが定義されており、その中でfirst_coroutineを呼び出していることがわかります。つまり、second_coroutineは、first_coroutineを呼び出すことで、処理を続けることができるようになっています。

このように、asynicoでは、コルーチンをチェーンとしてつなげることで、複雑な非同期処理を簡単かつ柔軟に実現することができます。

サーバークライアントのストリーム通信

サーバークライアント間でのストリーム通信には、通常、TCPソケットを使用します。asynicoを使用してサーバーとクライアントの両方を実装することができます。

以下は、asynicoを使用してサーバーとクライアントのストリーム通信を実現する簡単な例です。

まず、サーバーコードを見てみましょう。サーバーは、クライアントからの接続を待ち、接続されるとメッセージを受信して、そのメッセージに応答する単純なエコーサーバーです。

import asyncio

async def handle_echo(reader, writer):
while True:
data = await reader.read(100)
message = data.decode()
addr = writer.get_extra_info('peername')
print(f"Received {message!r} from {addr!r}")

writer.write(data)
await writer.drain()

async def main():
server = await asyncio.start_server(handle_echo, '127.0.0.1', 8888)

async with server:
await server.serve_forever()

asyncio.run(main())

次に、クライアントコードを見てみましょう。クライアントは、サーバーに接続して、メッセージを送信し、サーバーからの応答を受信します。

import asyncio

async def tcp_echo_client(message):
reader, writer = await asyncio.open_connection('127.0.0.1', 8888)

writer.write(message.encode())
await writer.drain()

data = await reader.read(100)
print(f"Received: {data.decode()!r}")

writer.close()
await writer.wait_closed()

asyncio.run(tcp_echo_client("Hello, World!"))

ネイティブコルーチンとジェネレーターベースのtypes.coroutine

Pythonのasyncioモジュールでは、非同期処理を実現するために、ネイティブコルーチン（native coroutine）とジェネレーターベースのコルーチン（generator-based coroutine）の2種類が用意されています。

ネイティブコルーチンは、async defという構文で定義されます。ネイティブコルーチンは、非同期処理の途中でawaitキーワードを使って他のコルーチンを呼び出すことができます。

ジェネレーターベースのコルーチンは、Python 2.5で導入されたtypes.coroutineデコレーターを使って、ジェネレータ関数をコルーチンに変換することができます。ジェネレーターベースのコルーチンは、Python 3.5以降でも引き続き利用可能ですが、ネイティブコルーチンと比べると少し使い勝手が悪いです。

以下は、ネイティブコルーチンとジェネレーターベースのコルーチンの例です。

import asyncio
import types

# ネイティブコルーチンの例
async def coro():
print('start')
await asyncio.sleep(1)
print('end')

# ジェネレーターベースのコルーチンの例
@types.coroutine
def coro():
print('start')
yield from asyncio.sleep(1)
print('end')

import asyncio

class AwaitableClass:
async def get_result(self):
await asyncio.sleep(2)
return 42

def __await__(self):
return self.get_result().__await__()

async def main():
result = await AwaitableClass()
print(result)

asyncio.run(main())

上記のコードは、2秒待機した後に42を出力します。__await__メソッドは、get_result()メソッドを呼び出して、非同期的に実行されます。その後、get_result()メソッドは、await asyncio.sleep(2)文によって2秒間待機し、42を返します。最後に、result変数には42が設定され、出力されます。

Awaitable Classは、コルーチンやタスクを作成する場合に便利です。__await__メソッドを使用して、非同期的に待機可能なオブジェクトを作成することができます。これにより、簡単に非同期処理を実現することができます。

async for文と__anext__

async for文と__anext__は、非同期イテレーションをサポートするための機能です。

async for文は、通常のfor文と同じようにループを回すことができますが、awaitキーワードを使用して非同期的に要素を取得します。

__anext__は、イテレーターによって実装され、async for文がawaitキーワードを使用して非同期的に要素を取得するために呼び出されます。この特殊メソッドは、次の要素を生成するために非同期処理を実行する必要があるため、コルーチンである必要があります。

以下は、async for文と__anext__を使用した例です。

import asyncio

class AsyncRange:
def __init__(self, start, stop, step=1):
self.i = start
self.stop = stop
self.step = step

async def __anext__(self):
if self.i < self.stop:
i = self.i
self.i += self.step
return i
else:
raise StopAsyncIteration

async def main():
async for i in AsyncRange(0, 10, 2):
print(i)

asyncio.run(main())

この例では、AsyncRangeというクラスを作成しています。このクラスは、__anext__メソッドを実装しており、非同期処理を使用して0から始まり、2ずつ増加し、10未満の範囲の値を生成するイテレーターを表します。main関数では、async for文を使用してイテレーターから要素を取得し、それを出力します。

このように、async for文と__anext__を使用することで、非同期処理でイテレーションを行うことができます。

async with文と__aenter__と__aexist__

async with文は、Pythonの非同期処理において、リソースの取得や解放のために使用されます。通常、同期処理におけるwith文は、__enter__と__exit__の特殊メソッドを定義して、リソースの取得と解放を処理します。しかし、非同期処理においては、これらの特殊メソッドを非同期的に行うために、async with文が導入されました。

async with文を使用する場合、クラスは__aenter__と__aexit__という特殊メソッドを実装する必要があります。__aenter__はasync withブロックに入る前に実行され、リソースを取得するために使用されます。__aexit__はasync withブロックを抜けるときに実行され、リソースを解放するために使用されます。

async withブロックは、async withによって返されたオブジェクトがAsyncContextManagerプロトコルを実装している場合に使用できます。AsyncContextManagerプロトコルは、__aenter__と__aexit__を定義するために使用されます。

以下は、async with文を使用した例です。

import asyncio

class MyResource:
async def __aenter__(self):
print('MyResource: acquiring resource')
return self

async def __aexit__(self, exc_type, exc, tb):
print('MyResource: releasing resource')

async def my_coroutine():
async with MyResource():
print('Coroutine: using resource')

asyncio.run(my_coroutine())

上記の例では、MyResourceクラスが__aenter__と__aexit__を実装しています。my_coroutine関数では、async with文を使用して、MyResourceオブジェクトを取得しています。MyResourceオブジェクトが取得されると、Coroutine: using resourceが出力されます。my_coroutine関数が完了すると、MyResource: releasing resourceが出力されます。

async withブロックを抜けるときに__aexit__が呼び出されることに注意してください。__aexit__は、async withブロック内で例外が発生した場合に、例外の型、値、およびトレースバックを受け取ります。したがって、__aexit__は、例外に対する適切な処理を行うことができます。

websocketsのご紹介

WebSocketsとは、双方向通信を行うためのプロトコルです。WebSocketsを使うことで、サーバーとクライアント間でリアルタイムにデータをやり取りすることができます。

従来のHTTPリクエストには、クライアントからサーバーへのリクエストが送られ、サーバーはそれに対してレスポンスを返すという形式でした。つまり、クライアントからリクエストがなければサーバー側からの通知はできないという制限があります。

しかし、WebSocketsを使うことで、サーバー側からの通知を受け取ることができます。これにより、リアルタイムな情報の受信や送信が可能になります。

WebSocketsは、JavaScriptやPythonなどのプログラミング言語で利用できます。WebSocketsの仕様は、W3Cによって策定されています。

blocking関数を並列に実行する場合

blocking関数を並列に実行する場合、複数の方法があります。いくつかの一般的な方法を以下に示します。

• スレッドを使用する Pythonには、threadingモジュールを使用してスレッドを作成できます。複数のスレッドを使用して、blocking関数を並列に実行することができます。ただし、スレッドを使用する場合、GILのためにPythonの並列処理が制限される可能性があります。
• プロセスを使用する Pythonには、multiprocessingモジュールを使用してプロセスを作成できます。複数のプロセスを使用して、blocking関数を並列に実行することができます。プロセスを使用すると、GILの影響を受けないため、Pythonの並列処理が改善されます。
• asyncioを使用する asyncioは、イベントループを使用して非同期I/O処理を実現するPythonの標準ライブラリです。非同期I/O処理は、シングルスレッドで実行されるため、GILの問題を回避できます。asyncioを使用して、複数のブロッキング関数を並列に実行できます。
• concurrent.futuresを使用する concurrent.futuresモジュールは、非同期実行を行うための高水準のインターフェースを提供するPythonの標準ライブラリです。concurrent.futuresモジュールを使用して、複数のブロッキング関数を並列に実行できます。内部的には、スレッドまたはプロセスを使用して、ブロッキング関数を実行します。

asynico.subprocess

asyncio.subprocessは、非同期でサブプロセスを実行するための機能を提供します。このモジュールを使用することで、Pythonで外部プログラムを起動して、その入出力ストリームを取得することができます。

asyncio.subprocessには、create_subprocess_execやcreate_subprocess_shellなどの関数があり、それらを使用することで、新しいサブプロセスを作成できます。これらの関数は、サブプロセスを作成するとともに、Processオブジェクトを返します。Processオブジェクトには、サブプロセスの入出力を管理するためのストリーム（stdin、stdout、stderr）が含まれています。

Processオブジェクトには、communicateメソッドがあります。このメソッドを使用することで、サブプロセスが終了するまで待機し、サブプロセスからの出力を取得できます。communicateメソッドは、stdinに書き込むデータを指定することができ、stdoutとstderrからの出力をバイト列として返します。また、stdin、stdout、stderrにアクセスするためのStreamWriterおよびStreamReaderオブジェクトを取得することができます。これらのオブジェクトを使用することで、非同期で入出力ストリームを処理することができます。

以下は、asyncio.subprocessを使用して、echoコマンドを実行し、その出力を取得する例です。

import asyncio

async def run_echo():
process = await asyncio.create_subprocess_exec(
'echo', 'hello', stdout=asyncio.subprocess.PIPE)
stdout, _ = await process.communicate()
return stdout.decode()

result = asyncio.run(run_echo())
print(result)

https://medical-science-labo.jp/python_advance13/