Python 코루틴 마스터: 강력한 동시 앱을 위한 사용자 지정 비동기 도구 만들기

Python의 코루틴은 비동기 코드 작성을 위한 강력한 도구입니다. 동시 작업을 처리하는 방식을 혁신하여 확장 가능하고 효율적인 애플리케이션을 더 쉽게 구축할 수 있게 되었습니다. 저는 코루틴 작업에 많은 시간을 보냈으며 사용자 정의 비동기 프리미티브 생성에 대한 몇 가지 통찰력을 공유하게 되어 기쁩니다.

기본부터 시작하겠습니다. 코루틴은 일시 중지하고 다시 시작할 수 있는 특수 기능으로 협력적인 멀티태스킹을 가능하게 합니다. 이는 Python의 async/await 구문의 기초입니다. 코루틴을 정의하면 본질적으로 이벤트 루프에 제어권을 넘겨주고 다른 작업을 실행할 수 있는 함수를 생성하게 됩니다.

사용자 정의 대기 가능 객체를 생성하려면 await 메서드를 구현해야 합니다. 이 메서드는 반복자를 반환해야 합니다. 다음은 간단한 예입니다.

class CustomAwaitable:
    def __init__(self, value):
        self.value = value

    def __await__(self):
        yield
        return self.value

async def use_custom_awaitable():
    result = await CustomAwaitable(42)
    print(result)  # Output: 42

이 CustomAwaitable 클래스는 내장된 어웨이터블과 마찬가지로 wait 키워드와 함께 사용할 수 있습니다. 기다렸다가 제어권을 한 번 양보한 다음 해당 값을 반환합니다.

하지만 더 복잡한 비동기 기본 요소를 만들고 싶다면 어떻게 해야 할까요? 사용자 정의 세마포어 구현을 살펴보겠습니다. 세마포어는 여러 코루틴의 공유 리소스에 대한 액세스를 제어하는 ​​데 사용됩니다.

import asyncio

class CustomSemaphore:
    def __init__(self, value=1):
        self._value = value
        self._waiters = []

    async def acquire(self):
        while self._value 



<p>이 CustomSemaphore 클래스는 비동기 컨텍스트 관리자 프로토콜(<strong>aenter</strong> 및 <strong>aexit</strong>)뿐만 아니라 획득 및 해제 메서드를 구현합니다. 최대 2개의 코루틴이 세마포어를 동시에 획득할 수 있습니다.</p>

<p>이제 효율적인 이벤트 루프를 만드는 방법에 대해 이야기해 보겠습니다. Python의 asyncio는 강력한 이벤트 루프 구현을 제공하지만 사용자 정의 구현이 필요한 경우가 있을 수 있습니다. 다음은 맞춤 이벤트 루프의 기본 예입니다.<br>
</p>

<pre class="brush:php;toolbar:false">import time
from collections import deque

class CustomEventLoop:
    def __init__(self):
        self._ready = deque()
        self._stopping = False

    def call_soon(self, callback, *args):
        self._ready.append((callback, args))

    def run_forever(self):
        while not self._stopping:
            self._run_once()

    def _run_once(self):
        ntodo = len(self._ready)
        for _ in range(ntodo):
            callback, args = self._ready.popleft()
            callback(*args)

    def stop(self):
        self._stopping = True

    def run_until_complete(self, coro):
        def _done_callback(fut):
            self.stop()

        task = self.create_task(coro)
        task.add_done_callback(_done_callback)
        self.run_forever()
        return task.result()

    def create_task(self, coro):
        task = Task(coro, self)
        self.call_soon(task._step)
        return task

class Task:
    def __init__(self, coro, loop):
        self._coro = coro
        self._loop = loop
        self._done = False
        self._result = None
        self._callbacks = []

    def _step(self):
        try:
            if self._done:
                return
            result = self._coro.send(None)
            if isinstance(result, SleepHandle):
                result._task = self
                self._loop.call_soon(result._wake_up)
            else:
                self._loop.call_soon(self._step)
        except StopIteration as e:
            self.set_result(e.value)

    def set_result(self, result):
        self._result = result
        self._done = True
        for callback in self._callbacks:
            self._loop.call_soon(callback, self)

    def add_done_callback(self, callback):
        if self._done:
            self._loop.call_soon(callback, self)
        else:
            self._callbacks.append(callback)

    def result(self):
        if not self._done:
            raise RuntimeError('Task is not done')
        return self._result

class SleepHandle:
    def __init__(self, duration):
        self._duration = duration
        self._task = None
        self._start_time = time.time()

    def _wake_up(self):
        if time.time() - self._start_time >= self._duration:
            self._task._loop.call_soon(self._task._step)
        else:
            self._task._loop.call_soon(self._wake_up)

async def sleep(duration):
    return SleepHandle(duration)

async def example():
    print("Start")
    await sleep(1)
    print("After 1 second")
    await sleep(2)
    print("After 2 more seconds")
    return "Done"

loop = CustomEventLoop()
result = loop.run_until_complete(example())
print(result)

이 맞춤 이벤트 루프는 작업 실행, 코루틴 처리, 간단한 절전 기능과 같은 기본 기능을 구현합니다. Python의 내장 이벤트 루프만큼 기능이 풍부하지는 않지만 핵심 개념을 보여줍니다.

비동기 코드 작성 시 어려운 점 중 하나는 작업 우선순위를 관리하는 것입니다. Python의 asyncio는 작업에 대한 기본 우선순위 대기열을 제공하지 않지만 다음과 같이 자체적으로 구현할 수 있습니다.

import asyncio
import heapq

class PriorityEventLoop(asyncio.AbstractEventLoop):
    def __init__(self):
        self._ready = []
        self._stopping = False
        self._clock = 0

    def call_at(self, when, callback, *args, context=None):
        handle = asyncio.Handle(callback, args, self, context)
        heapq.heappush(self._ready, (when, handle))
        return handle

    def call_later(self, delay, callback, *args, context=None):
        return self.call_at(self._clock + delay, callback, *args, context=context)

    def call_soon(self, callback, *args, context=None):
        return self.call_at(self._clock, callback, *args, context=context)

    def time(self):
        return self._clock

    def stop(self):
        self._stopping = True

    def is_running(self):
        return not self._stopping

    def run_forever(self):
        while self._ready and not self._stopping:
            self._run_once()

    def _run_once(self):
        if not self._ready:
            return
        when, handle = heapq.heappop(self._ready)
        self._clock = when
        handle._run()

    def create_task(self, coro):
        return asyncio.Task(coro, loop=self)

    def run_until_complete(self, future):
        asyncio.futures._chain_future(future, self.create_future())
        self.run_forever()
        if not future.done():
            raise RuntimeError('Event loop stopped before Future completed.')
        return future.result()

    def create_future(self):
        return asyncio.Future(loop=self)

async def low_priority_task():
    print("Low priority task started")
    await asyncio.sleep(2)
    print("Low priority task finished")

async def high_priority_task():
    print("High priority task started")
    await asyncio.sleep(1)
    print("High priority task finished")

async def main():
    loop = asyncio.get_event_loop()
    loop.call_later(0.1, loop.create_task, low_priority_task())
    loop.call_later(0, loop.create_task, high_priority_task())
    await asyncio.sleep(3)

asyncio.run(main())

이 PriorityEventLoop는 힙 대기열을 사용하여 예약된 실행 시간에 따라 작업을 관리합니다. 지연 시간이 다른 작업을 예약하여 우선순위를 지정할 수 있습니다.

취소를 적절하게 처리하는 것은 코루틴 작업의 또 다른 중요한 측면입니다. 취소 가능한 작업을 구현하는 방법의 예는 다음과 같습니다.

import asyncio

async def cancellable_operation():
    try:
        print("Operation started")
        await asyncio.sleep(5)
        print("Operation completed")
    except asyncio.CancelledError:
        print("Operation was cancelled")
        # Perform any necessary cleanup
        raise  # Re-raise the CancelledError

async def main():
    task = asyncio.create_task(cancellable_operation())
    await asyncio.sleep(2)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Main: task was cancelled")

asyncio.run(main())

이 예에서 cancelled_option은 CancelledError를 포착하고 필요한 정리를 수행한 다음 예외를 다시 발생시킵니다. 이를 통해 취소 상태를 계속 전파하면서 취소를 정상적으로 처리할 수 있습니다.

맞춤형 비동기 반복기 구현을 살펴보겠습니다. 이는 비동기적으로 반복될 수 있는 시퀀스를 생성하는 데 유용합니다.

class CustomAwaitable:
    def __init__(self, value):
        self.value = value

    def __await__(self):
        yield
        return self.value

async def use_custom_awaitable():
    result = await CustomAwaitable(42)
    print(result)  # Output: 42

이 AsyncRange 클래스는 비동기 반복자 프로토콜을 구현하여 비동기 for 루프에서 사용할 수 있도록 합니다.

마지막으로 사용자 정의 비동기 컨텍스트 관리자 구현을 살펴보겠습니다. 이는 비동기적으로 획득하고 해제해야 하는 리소스를 관리하는 데 유용합니다.

import asyncio

class CustomSemaphore:
    def __init__(self, value=1):
        self._value = value
        self._waiters = []

    async def acquire(self):
        while self._value 



<p>이 AsyncResource 클래스는 <strong>aenter</strong> 및 <strong>aexit</strong> 메소드를 구현하여 async with 문과 함께 사용할 수 있습니다.</p>

<p>결론적으로 Python의 코루틴 시스템은 사용자 정의 비동기 프리미티브를 구축하기 위한 강력한 기반을 제공합니다. 기본 메커니즘과 프로토콜을 이해함으로써 특정 비동기 문제에 대한 맞춤형 솔루션을 만들고, 복잡한 동시 시나리오에서 성능을 최적화하고, Python의 비동기 기능을 확장할 수 있습니다. 이러한 사용자 정의 구현은 학습 및 특정 사용 사례에 적합하지만 Python의 내장 asyncio 라이브러리는 고도로 최적화되어 있으며 대부분의 시나리오에서 사용해야 합니다. 즐거운 코딩하세요!</p>


<hr>

<h2>
  
  
  우리의 창조물
</h2>

<p>저희 창작물을 꼭 확인해 보세요.</p>

<p><strong>인베스터 센트럴</strong> | <strong>스마트리빙</strong> | <strong>시대와 메아리</strong> | <strong>수수께끼의 미스터리</strong> | <strong>힌두트바</strong> | <strong>엘리트 개발자</strong> | <strong>JS 학교</strong></p>


<hr>

<h3>
  
  
  우리는 중간에 있습니다
</h3>

<p><strong>테크 코알라 인사이트</strong> | <strong>Epochs & Echoes World</strong> | <strong>투자자중앙매체</strong> | <strong>수수께끼 미스터리 매체</strong> | <strong>과학과 신기원 매체</strong> | <strong>현대 힌두트바</strong></p>


          

            
        

위 내용은 Python 코루틴 마스터: 강력한 동시 앱을 위한 사용자 지정 비동기 도구 만들기의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!