Ачааллж байна...
Өмнөх хоёр хичээлд бид asyncio-гийн хүчийг харсан: зуун мянган coroutine хоёр секунд хүрэхгүй хугацаанд, TaskGroup алдааг цэвэрхэн удирдана, timeout мөнхийн хүлээлтээс хамгаална. Энэ хичээл тэр хүчийг бодит ертөнцтэй уулзуулна. Бодит ертөнц хоёр хязгаартай.
Эхний хязгаар бол гадны систем. Хичээл 13.2-т бид серверүүд хэт олон зэрэг хүсэлтийг “429 Too Many Requests” гэж татгалздагийг тэмдэглэсэн. asyncio-д энэ асуудал улам хурцадна: thread-ийн pool ядаж max_workers-ээр хязгаарлагддаг байсан, харин мянган task бүгд яг нэг агшинд сервер рүү дайрна. Бидэнд "нэг удаад тавааc илүүгүй" гэж хэлэх хэрэгсэл хэрэгтэй: Semaphore.
Хоёр дахь хязгаар бол асинхрон бус код. Хичээл 14.1-ийн хамгийн чухал дүрэм: loop-ийг бүү блокло. Гэвч Python-ий сангуудын ихэнх нь asyncio-гоос өмнө бичигдсэн бөгөөд синхрон: тэд await хийдэггүй, thread-ийг зогсоодог. Хуучин мэдээллийн сангийн драйвер, файл боловсруулах сан, Хичээл 13.1-ийн crunch шиг тооцоолол. Тэдгээрийг asyncio програмаас хэрхэн дуудах вэ? Хариулт нь Бүлэг 13-ын хэрэгслүүд, thread ба process, asyncio-гийн loop-тэй холбогдох нь: asyncio.to_thread ба run_in_executor. Энэ хичээл Түвшин 5-ын гурван хэрэгслийг нэг програмд нэгтгэж, хэзээ алийг нь хэрэглэхийг хэмжилтээр баталгаажуулна.
Олон нийтийн усан бассейныг төсөөлөөд үзээрэй. Бассейн дөрвөн замтай. Хаалган дээр дөрвөн бугуйвч өлгөөстэй. Усанд орох хүн бугуйвч авч орно, гарахдаа буцааж өлгөнө. Бугуйвч дууссан бол дараагийн хүн хаалган дээр хүлээнэ. Хэн ч түүнийг хүчээр зогсоохгүй, бүртгэл хөтлөх ажилтан ч хэрэггүй: бугуйвчны тоо өөрөө хязгаар. Энэ бол Semaphore: тоолдог түгжээ. Хичээл 13.2-ын Lock бол ганц бугуйвчтай semaphore.
Одоо асинхрон бус кодын зүйрлэл. Симултан тоглолтын мастерыг санаарай. Нэг ширээний өрсөлдөгч "миний нүүдлийг шинжлэхэд нарийн тооцоо хэрэгтэй, хорин минут" гэж хэлэв. Мастер тэнд зогсож тооцоолбол бусад арван есөн ширээ хорин минут хүлээнэ. Ухаалаг мастер тэр тооцоог туслахад өгнө: туслах хажуугийн өрөөнд тооцоолж байх хооронд мастер бусад ширээгээр явна. Туслах дуусмагц мастерт үр дүнг өгнө. Энэ бол to_thread: блоклох ажлыг тусдаа thread-д шилжүүлж, loop-ийг чөлөөтэй байлгах. Хэрэв тооцоо маш хүнд бөгөөд нэг туслах хангалтгүй бол, Хичээл 13.3-ын GIL-ийг санаарай, тусдаа салбарын туслахуудад өгнө: process.
Хорин хуудас, тус бүр нэг секунд. Нэг удаад хэдэн хүсэлт зэрэг явж байгааг тоолох жижиг класс нэмье:
python
import asyncio
import time
class Tracker:
def __init__(self):
self.current = 0
self.peak = 0
async def polite_fetch(name, semaphore, tracker):
async with semaphore:
tracker.current += 1
tracker.peak = max(tracker.peak, tracker.current)
await asyncio.sleep(1.0)
tracker.current -= 1
return f"{name}: татагдлаа"
async def main(limit):
semaphore = asyncio.Semaphore(limit)
tracker = Tracker()
start = time.perf_counter()
async with asyncio.TaskGroup() as group:
for i in range(20):
group.create_task(polite_fetch(f"хуудас-{i}", semaphore, tracker))
elapsed = time.perf_counter() - start
print(f"Хязгаар {limit:>2}: {elapsed:5.2f} с, нэг зэрэг хамгийн ихдээ {tracker.peak}")
for limit in (1, 5, 20):
asyncio.run(main(limit))Хязгаар 1: 20.02 с, нэг зэрэг хамгийн ихдээ 1
Хязгаар 5: 4.01 с, нэг зэрэг хамгийн ихдээ 5
Хязгаар 20: 1.00 с, нэг зэрэг хамгийн ихдээ 20asyncio.Semaphore(limit) нь limit тооны "бугуйвчтай" түгжээ. async with semaphore блок руу орохдоо нэг бугуйвч авна, гарахдаа, алдаатай ч, буцааж өгнө. Бугуйвч дууссан бол coroutine тэр мөрөнд await хийж, ээлжээ өгч хүлээнэ: loop-ийг блоклохгүй. Хорин task бүгд нэг агшинд үүссэн, гэвч peak тоо хязгаарыг хэзээ ч хэтрээгүй. Хугацаа Хичээл 13.2-ын thread pool-тэй яг ижил хэв маягтай: хязгаар 5 бол дөрвөн давалгаа, дөрвөн секунд.
Tracker-ийн тоолуурт Lock хэрэглээгүйг анзаараарай. Хичээл 14.1-ийн гол санаа энд ажиллаж байна: current += 1 ба peak-ийг шинэчлэх хооронд await байхгүй тул өөр coroutine тэдний дунд орж ирэх боломжгүй. Thread-тэй ижил код race condition-тэй байх байсан.
Энэ бүтцийн гоё тал: хязгаар ажлын дотор, ажлыг эхлүүлэх газарт биш. TaskGroup хорин task-ийг шууд үүсгэж, хязгаарын асуудлыг тоохгүй. Polite_fetch хаанаас ч дуудагдсан бай, semaphore-ийг дагана. Мөн semaphore-ийг main дотор, ажиллаж буй loop дотор үүсгэж, аргументаар дамжуулсан: Түгээмэл алдаанууд хэсэгт яагаад гэдгийг харна.
Шударга тэмдэглэл: Semaphore зэрэг хүсэлтийн тоог хязгаарладаг, секундэд хэдэн хүсэлт гэдгийг биш. Олон API "минутад 60 хүсэлт" гэх мэт хурдны хязгаартай байдаг. Тэдгээрт semaphore дээр нэмж хүсэлт бүрийн хооронд asyncio.sleep хийх, эсвэл тусгай хязгаарлагч хэрэгтэй. Эхлээд таны хэрэглэх API-ийн баримт бичгийг уншиж, ямар хязгаар байгааг мэд.
Одоо хоёр дахь асуудал. Синхрон сан дуудаж байгааг time.sleep-ээр загварчилъя. Мөн loop амьд байгаа эсэхийг хэмжих хэрэгсэл хэрэгтэй: тогтмол "зүрх цохилдог" coroutine. Loop чөлөөтэй бол зүрх жигд цохилно, блоклогдвол зогсоно:
python
def blocking_fetch(name, seconds=1.0):
time.sleep(seconds)
return f"{name}: татагдлаа"
async def heartbeat(stop):
beats = 0
while not stop.is_set():
await asyncio.sleep(0.25)
beats += 1
return beats
async def measure(label, work):
stop = asyncio.Event()
beat_task = asyncio.create_task(heartbeat(stop))
start = time.perf_counter()
await work()
elapsed = time.perf_counter() - start
stop.set()
beats = await beat_task
print(f"{label:<12} {elapsed:.2f} с, зүрхний цохилт: {beats}")blocking_fetch бол энгийн def функц: синхрон сангийн загвар. asyncio.Event бол coroutine-уудын хоорондох дохио: stop.set() дуудагдмагц is_set() True буцаана. heartbeat хагас секундийн дөрөвний нэг тутамд нэг удаа цохилно: нэг секундэд дөрвөн цохилт хүлээгдэнэ. Одоо хоёр арга:
python
async def call_directly(name):
return blocking_fetch(name)
async def direct_work():
await asyncio.gather(*(call_directly(f"хуудас-{i}") for i in range(3)))
async def thread_work():
await asyncio.gather(*(asyncio.to_thread(blocking_fetch, f"хуудас-{i}") for i in range(3)))
async def main():
await measure("Шууд", direct_work)
await measure("to_thread", thread_work)
asyncio.run(main())Шууд 3.00 с, зүрхний цохилт: 1
to_thread 1.00 с, зүрхний цохилт: 4Шууд дуудах нь Хичээл 14.1-ийн блоклох жишээ: гурван секунд, бүх coroutine дараалсан. Зүрх гурван секундэд ганц удаа л цохилсон: loop бүхэлдээ хөлдсөн. Хэрэглэгчийн интерфейстэй програм, эсвэл бусад хэрэглэгчдэд үйлчилж буй сервер энэ гурван секундэд бүрэн хариугүй болно.
asyncio.to_thread(func, *args) нь энгийн функцийг тусдаа thread дээр ажиллуулж, түүнийг await хийж болох объект болгоно. Гурван блоклох дуудлага гурван thread дээр зэрэг явсан: нэг секунд. Хамгийн чухал нь зүрх жигд цохилсон: дөрвөн цохилт, хүлээгдсэн хэмжээгээр. Loop тэр хооронд чөлөөтэй байж, бусад coroutine-д үйлчилсэн. Мастер тооцоог туслахад өгсөн.
Энэ бол асинхрон ба синхрон ертөнцийн гүүр. to_thread-ийн дотор Хичээл 13.2-ын ThreadPoolExecutor ажилладаг: loop-ийн анхдагч thread pool. Тиймээс түүний хязгаар энд ч хүчинтэй: pool-ийн хэмжээ цөмийн тоо дээр дөрөв, 32-оос ихгүй. Хэрэв зуун to_thread дуудлагыг зэрэг өгвөл тэд pool-ийн хэмжээгээр давалгаалж явна. Найман цөмтэй компьютер дээр зуун нэг секундийн блоклох дуудлага нэг секунд биш, ойролцоогоор найман секунд зарцуулна. Энэ нь ихэвчлэн сайн хэрэг, гадны системийг хэт ачаалахгүй, гэхдээ coroutine-уудын хязгааргүй мэт хэмжээг to_thread-ээс бүү хүлээ.
to_thread блоклох I/O-д тохирно: thread хүлээж байхдаа GIL-ийг сулладаг. Харин Хичээл 13.1-ийн crunch-ийг to_thread-д өгвөл loop чөлөөтэй үлдэнэ, гэвч тооцоолол хурдлахгүй: Хичээл 13.3-ын GIL. CPU-bound ажилд process хэрэгтэй. Loop-ийг process pool-тэй холбох хэрэгсэл бол loop.run_in_executor:
python
from concurrent.futures import ProcessPoolExecutor
def crunch(n=5_000_000):
total = 0
for i in range(n):
total += i * i
return total
async def main():
loop = asyncio.get_running_loop()
start = time.perf_counter()
await asyncio.gather(*(asyncio.to_thread(crunch) for _ in range(4)))
print(f"to_thread x4: {time.perf_counter() - start:.2f} с")
start = time.perf_counter()
with ProcessPoolExecutor() as pool:
await asyncio.gather(*(loop.run_in_executor(pool, crunch) for _ in range(4)))
print(f"process x4: {time.perf_counter() - start:.2f} с")
if __name__ == "__main__":
asyncio.run(main())to_thread x4: 1.34 с
process x4: 0.41 сХичээл 13.3-ын хэмжилт асинхрон хэлбэрээр давтагдсан: thread тооцооллыг хурдасгаагүй, process гурав дахин хурдасгасан. asyncio.get_running_loop() одоо ажиллаж буй loop-ийг өгнө. loop.run_in_executor(pool, func, *args) нь func-ийг pool дотор ажиллуулж, await хийж болох объект буцаана. Анхны аргумент нь None бол loop-ийн анхдагч thread pool хэрэглэгдэнэ: to_thread яг үүнийг хийдэг. if __name__ == "__main__": хамгаалалт энд ч process-ийн шаардлагаар заавал, Хичээл 13.3-ын адил.
Нэг тэмдэглэл: pool-ийг with-ээр нээж хаах нь энэ жишээнд тохиромжтой. Бодит сервер програмд process эхлүүлэх зардлыг давтахгүйн тулд pool-ийг програмын эхэнд нэг удаа үүсгэж, дуустал хадгалдаг.
Түвшин 5-ын бүх хэрэгслийг нэг газар цуглуулъя.
asyncio coroutine бол async сан бүхий I/O-bound ажлын анхны сонголт: олон мянган зэрэг хүлээлт, хөнгөн, солилтын цэг кодонд харагддаг. Гадны системийн ачааллыг Semaphore-оор хязгаарла.
to_thread бол async програм дотроос синхрон, блоклох I/O дуудах гүүр: async хувилбаргүй сан, файлын системийн том үйлдэл. Loop-ийг чөлөөтэй байлгана, хэмжээ нь thread pool-оор хязгаарлагдана.
run_in_executor ба ProcessPoolExecutor бол async програм дотроос CPU-bound тооцоолол хийх зам: тооцоолол цөмүүд дээр зэрэг явж, loop хариутай хэвээр үлдэнэ.
ThreadPoolExecutor ба ProcessPoolExecutor шууд, asyncio-гүйгээр: програм бүхэлдээ синхрон бол, зэрэгцээ хэсэг нь цөөн бөгөөд тодорхой бол. Хичээл 13.2-13.3-ын кодыг asyncio руу хөрвүүлэх шаардлага байхгүй: хэрэв ажил хийгдэж байгаа бол энгийн хэвээр үлдээ.
Асуулт бүр Хичээл 13.1-ийн хоёр цагийн хэмжилтээс эхэлнэ: хүлээлт үү, тооцоолол уу?
Хүсвэл polite_fetch-д 10 хувийн магадлалтайгаар ConnectionError шидэх нөхцөл нэмж, Хичээл 14.2-ын TaskGroup-ийн цуцлалт semaphore-ийн бугуйвчуудыг зөв буцааж байгааг шалгаарай: алдааны дараа peak тоо болон current тоо хэвийн үү? Сонирхвол measure-ийг Хичээл 12.4-ийн импортлогчийн import_orders дуудлагад хэрэглэж, түүнийг шууд ба to_thread-ээр дуудахад зүрхний цохилт хэрхэн өөрчлөгдөхийг харж болно. Мөн to_thread-ийн давалгааг өөрийн нүдээр харахын тулд 3-ын оронд 50 blocking_fetch өгч хэмжих нь сонирхолтой: таны компьютерийн анхдагч pool хэдэн thread-тэй вэ? Эдгээр бүгд санал бөгөөд дараагийн хичээлүүд үүнийг хийсэн гэж үзэхгүй.
RuntimeError: <asyncio.locks.Semaphore object at 0x...> is bound to a different event loop. “Semaphore өөр event loop-той холбогдсон” гэсэн утгатай. Semaphore-ийг модулийн түвшинд нэг удаа үүсгээд, asyncio.run-ийг хэд хэдэн удаа дуудахад гарна: энэ хичээлийн хэмжилтийн давталт шиг. Semaphore анх хэрэглэгдэх үедээ тэр үеийн loop-той холбогддог, дараагийн asyncio.run шинэ loop үүсгэдэг. Шийдэл: semaphore, Lock, Event зэрэг asyncio-гийн хэрэгслүүдийг coroutine дотор, ажиллаж буй loop дотор үүсгэж, аргументаар дамжуул.
Semaphore-ийг буруу газар тавих. Мессежгүй. async with semaphore: блок дотор create_task дуудвал semaphore task-уудыг үүсгэх алхмыг хязгаарлана: үүсгэх нь агшин зуурын тул бүх task бараг нэг дор эхэлж, хязгаар огт үйлчлэхгүй. Бугуйвчийг усанд орох хүн өөрөө авах ёстой: semaphore-ийг хязгаарлах гэж буй бодит ажлын эргэн тойронд, coroutine-ийн дотор тавь. Tracker-ийн peak шиг тоолуур нь хязгаар ажиллаж байгааг батлах хамгийн хялбар арга.
to_thread-д async функц өгөх. Мессеж нь RuntimeWarning: coroutine 'fetch' was never awaited. to_thread-ийн функц энгийн def байх ёстой. Хэрэв async def fetch-ийг өгвөл thread түүнийг дуудаж, coroutine объект л буцаана: Хичээл 14.1-ийн адил ажиллахгүй coroutine. Үр дүн нь coroutine объект, ажил хэзээ ч хийгдээгүй. Async функцийг шууд await хий, to_thread зөвхөн синхрон функцэд.
TypeError: run_in_executor() got an unexpected keyword argument ’n’. “run_in_executor() n гэсэн хүлээгдээгүй keyword аргумент авлаа” гэсэн утгатай. run_in_executor зөвхөн байрлалын аргументуудыг дамжуулдаг. Keyword аргумент хэрэгтэй бол Хичээл 3.3-ын functools.partial: loop.run_in_executor(pool, partial(crunch, n=1_000_000)). Хичээл 13.3-т хэлсэнчлэн partial объект process руу pickle хийгдэх боломжтой. to_thread харин keyword аргументыг шууд хүлээж авдаг.
CPU-bound ажлыг to_thread-д өгөөд хурдасна гэж хүлээх. Мессежгүй, хэмжилтээр харсан. to_thread loop-ийг чөлөөтэй байлгана, гэвч GIL-ийн улмаас тооцоолол хурдлахгүй. Хоёр зорилгыг ялга: “loop хариутай байх” бол to_thread хангалттай, "тооцоолол хурдан болох" бол process.
asyncio.Semaphore бол тоолдог түгжээ: async with semaphore блок руу зэрэг limit-ээс олон coroutine орохгүй, бусад нь loop-ийг блоклолгүй хүлээнэ. Хязгаарыг бодит ажлын эргэн тойронд, coroutine дотор тавь. Semaphore зэрэг хүсэлтийг хязгаарлана, секундэд хэдийг биш. asyncio-гийн хэрэгслүүдийг ажиллаж буй loop дотор үүсгэ, эс бөгөөс "өөр loop-той холбогдсон" алдаа гарна. Хоёр await-ийн хооронд tracker-ийн тоолуур шиг төлөвийг Lock-гүйгээр өөрчилж болно. asyncio.to_thread синхрон, блоклох функцийг тусдаа thread дээр ажиллуулж, await хийж болох болгоно: loop чөлөөтэй үлдэнэ. Зүрхний цохилтын хэмжилт үүнийг баталсан: шууд дуудахад гурван секундэд нэг цохилт, to_thread-ээр жигд дөрвөн цохилт. to_thread анхдагч thread pool ашигладаг тул түүний хэмжээгээр хязгаарлагдана. CPU-bound ажилд loop.run_in_executor-ийг ProcessPoolExecutor-тэй хамт хэрэглэнэ: тооцоолол цөмүүд дээр зэрэг, loop хариутай. Хэрэгслийн сонголт үргэлж хоёр цагийн хэмжилтээс эхэлнэ.
Ингээд Бүлэг 14 дууслаа. Бидний бүх asyncio жишээ asyncio.sleep-ээр сүлжээг загварчилсан. Дараагийн бүлэг жинхэнэ сүлжээ рүү шилжинэ. Мөн Түвшин 1-тэй гайхалтай холбоо бий болно: Бүлэг 1-2-т сурсан iterator ба generator-ийн asyncio хувилбарууд, async for ба async generator. Хичээл 2.3-ын pipeline-ийг санаарай: өгөгдөл үе шат дамжин урсдаг байсан. Async generator-ээр тэр pipeline-ийн үе шат бүр хүлээх чадвартай болно: сүлжээнээс ирэх өгөгдлийг ирэх тусам нь боловсруулах. Түвшин 5 нэгдсэн төсөлөөр дуусна.
Бүртгэлтэй болсноор энэ сургалтын бүх хичээлд хандах эрх авна.