coroutines in Python sind ein mächtiges Tool zum Schreiben von asynchronem Code. Sie haben revolutioniert, wie wir mit gleichzeitigen Vorgängen umgehen, und erleichtert es, skalierbare und effiziente Anwendungen aufzubauen. Ich habe viel Zeit damit verbracht, mit Coroutinen zu arbeiten, und ich freue mich, einige Einblicke in die Erstellung von benutzerdefinierten asynchronen Primitiven zu teilen.

Beginnen wir mit den Grundlagen. Coroutinen sind spezielle Funktionen, die übernommen und wieder aufgenommen werden können, was kooperativen Multitasking ermöglicht. Sie sind die Grundlage von Pythons Async/Aused Syntax. Wenn Sie eine Coroutine definieren, erstellen Sie im Wesentlichen eine Funktion, mit der die Ereignisschleife kontrolliert werden kann, sodass andere Aufgaben ausgeführt werden können.

Um ein benutzerdefiniertes erwartbares Objekt zu erstellen, müssen Sie die -Methode implementieren. Diese Methode sollte einen Iterator zurückgeben. Hier ist ein einfaches Beispiel:

Klasse CustomAwaitable: def __init __ (Selbst, Wert): self.Value = Wert Def __await __ (Selbst): Ertrag Return Self.Value Async def gebrauchs_custom_awaitable (): Ergebnis = Warten Sie anpaceAwaitable (42) Druck (Ergebnis) # Ausgabe: 42

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

Diese anpassbare Klasse kann mit dem Keyword für das Warten verwendet werden, genau wie integrierte Awaitables. Wenn es erwartet wird, gibt es einmal die Kontrolle und gibt dann seinen Wert zurück.

Aber was ist, wenn wir komplexere asynchrone Primitive schaffen wollen? Sehen wir uns die Implementierung eines benutzerdefinierten Semaphors an. Semaphoren werden verwendet, um den Zugriff auf eine gemeinsame Ressource durch mehrere Coroutinen zu steuern:

import asyncio Klassenmustern: def __init __ (self, value = 1): self._Value = Wert self._waiters = [] Async definitiv (Selbst): während self._Value 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 Diese Anpassungsemaphore -Klasse implementiert die Erwerbs- und Freigabemethoden sowie das Async Context Manager -Protokoll (

aenter und aexit ). Es ermöglicht maximal zwei Coroutinen, das Semaphor gleichzeitig zu erwerben.

Lassen Sie uns nun über das Erstellen effizienter Ereignisschleifen sprechen. Während Pythons Asyncio eine robuste Ereignisschleife implementiert, gibt es möglicherweise Fälle, in denen Sie eine benutzerdefinierte benötigen. Hier ist ein grundlegendes Beispiel für eine benutzerdefinierte Ereignisschleife:

Importzeit aus den Sammlungen importieren Deque Klasse CustomeEventLoop: def __init __ (selbst): self._ready = deque () self._sopping = false Def Call_SOon (Selbst, Rückruf, *args): self._ready.Append ((Rückruf, Argumente)) Def Run_forever (Selbst): Während nicht self._stoping: self._run_once () def _run_once (self): ntodo = len (self._ready) für _ im Bereich (Ntodo): Rückruf, args = self._ready.popleft () Rückruf (*args) def stopp (selbst): self._sopping = 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) Rückgabe Klassenaufgabe: Def __init __ (Selbst, Coro, Schleife): self._coro = coro self._loop = Loop self._done = false self._result = keine self._callbacks = [] def _step (self): versuchen: Wenn self._done: zurückkehren result = self._coro.send (keine) Wenn is is inance (result, sleepHandle): result._task = self self._loop.call_soon (result._wake_up) anders: self._loop.call_soon (self._step) Außer Stoppitation als e: self.set_result (E. value) Def set_result (Selbst, Ergebnis): self._result = Ergebnis self._done = true Für einen Rückruf in self._callbacks: self._loop.call_soon (Rückruf, Selbst) Def add_done_callback (Selbst, Rückruf): Wenn self._done: self._loop.call_soon (Rückruf, Selbst) anders: self._callbacks.append (Rückruf) Def Ergebnis (Selbst): Wenn nicht self._done: RunTimeError erhöhen ('Aufgabe ist nicht erledigt “) return self._result Klassenschlafhand: def __init __ (Selbst, Dauer): self._duration = dauer self._task = Keine self._start_time = time.time ()) Def _wake_Up (Selbst): if time.time () - self._start_time> = self._duration: self._task._loop.call_soon (self._task._step) anders: self._task._loop.call_soon (self._wake_up) Async defep Sleep (Dauer): Return SleepHandle (Dauer) Async def educt (): print ("start") Warten Sie den Schlaf (1) drucken ("nach 1 Sekunde") Warten Sie den Schlaf (2) print ("nach 2 weiteren Sekunden") zurück "fertig" Loop = CustomeEventLoop () result = Loop.run_until_complete (example ()) Druck (Ergebnis)

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

Diese benutzerdefinierte Ereignisschleife implementiert grundlegende Funktionen wie Ausführung von Aufgaben, Coroutinen und sogar eine einfache Schlaffunktion. Es ist nicht so merkmalreich wie Pythons integrierte Ereignisschleife, aber es demonstriert die Kernkonzepte.

Eine der Herausforderungen beim Schreiben von asynchronem Code ist die Verwaltung von Aufgabenprioritäten. Während Pythons Asyncio keine integrierten Priority-Warteschlangen für Aufgaben bereitstellt, können wir unsere eigenen implementieren:

import asyncio Heapq importieren Klasse PriorityEventLoop (asyncio.abstractEventLoop): def __init __ (selbst): self._ready = [] self._sopping = false self._clock = 0 Def call_at (self, wann, callback, *args, context = keine): Handle = asyncio.Handle (Rückruf, Argumentation, Selbst, Kontext) Heapq.Heappush (self._ready, (wenn, Handle)) Rückgang Def call_later (Selbst, Verzögerung, Rückruf, *args, context = keine): return self.call_at (self._clock delay, radback, *args, context = context) Def call_soon (self, callback, *args, context = keine): return self.call_at (self._clock, callback, *args, context = context) def Zeit (Selbst): Return self._clock def stopp (selbst): self._sopping = true def is_running (self): Rückgabe nicht self._Stoping Def Run_forever (Selbst): während self._ready und nicht self._stoping: self._run_once () def _run_once (self): Wenn nicht self._ready: zurückkehren Wenn, handle = heapq.heapop (self._ready) self._clock = wenn Handle._run () Def create_task (self, coro): asyncio.task zurückgeben (coro, Loop = self) Def Run_until_complete (Selbst, Zukunft): asyncio.futures._chain_future (Future, self.create_future ()) self.run_forever () Wenn nicht zukünftig.done (): RuntimeError erhöhen ('Event Loop, bevor die Zukunft abgeschlossen ist.') Return Future.Result () Def create_future (Selbst): asyncio.future zurückgeben (Loop = self) Async def Low_Priority_task (): drucken ("Aufgabe mit niedriger Priorität gestartet") Warten Sie Asyncio.sleep (2) print ("Task mit niedriger Priorität beendet") Async Def High_Priority_Task (): print ("Aufgabe mit hoher Priorität gestartet") Warten Sie Asyncio.sleep (1) print ("Aufgabe mit hoher Priorität abgeschlossen") 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 ()) Warten Sie Asyncio.sleep (3) asyncio.run (main ())

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

Diese PriorityEventLoop verwendet eine Heap -Warteschlange, um Aufgaben basierend auf ihrer geplanten Ausführungszeit zu verwalten. Sie können Prioritäten zuweisen, indem Sie Aufgaben mit unterschiedlichen Verzögerungen planen.

Die Anmeldung anmutig ist ein weiterer wichtiger Aspekt bei der Arbeit mit Coroutinen. Hier ist ein Beispiel für die Implementierung von bancellablen Aufgaben:

import asyncio Async Def Cancellable_operation (): versuchen: Druck ("Operation begonnen") Warten Sie Asyncio.sleep (5) drucken ("Betrieb abgeschlossen") Außer Asyncio.CancelledError: Druck ("Operation wurde storniert") # Führen Sie eine notwendige Reinigung durch Erhöhen Sie den CancelledError erneut. Async def Main (): task = asyncio.create_task (cancellable_operation ()) Warten Sie Asyncio.sleep (2) task.cancel () versuchen: Warten Sie auf die Aufgabe Außer Asyncio.CancelledError: print ("Main: Aufgabe wurde abgebrochen")) asyncio.run (main ())

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

In diesem Beispiel fängt die Cancellable_operation den CancelledError an, führt alle erforderlichen Aufräumarbeiten durch und rägt dann die Ausnahme wieder. Dies ermöglicht eine anmutige Umstellung der Stornierung und verbreitet den Stornierungsstatus weiterhin.

Lassen Sie uns die Implementierung benutzerdefinierter asynchronisierter Iteratoren untersuchen. Diese sind nützlich, um Sequenzen zu erstellen, die über asynchron iteriert werden können:

Klasse Asyncrange: Def __init __ (Selbst, Start, Stopp, Schritt = 1): self.start = start self.stop = stop self.step = Schritt Def __aiter __ (Selbst): Rückkehr selbst Async def __anext __ (Selbst): Wenn self.start> = self.stop: Stopasynciteration erhöhen value = self.start self.start = self.step Warten Rückgabewert Async def Main (): Async für i in Asyncrange (0, 5): drucken (i) asyncio.run (main ())

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

Diese Asyncrange -Klasse implementiert das Async -Iteratorprotokoll, sodass es in Async für Loops verwendet werden kann.

Schließlich schauen wir uns die Implementierung benutzerdefinierter asynchrischer Kontextmanager an. Diese sind nützlich, um Ressourcen zu verwalten, die asynchron erworben und veröffentlicht werden müssen:

Klasse Asyncresource: Async def __aenter __ (Selbst): print ("Ressource erwerben") Warten Rückkehr selbst Async def __aexit __ (self, exc_type, exc, tb): print ("Ressource veröffentlichen") Warten Async def Main (): Async mit Asyncresource () als Ressource: print ("Verwenden von Ressource") Warten Sie Asyncio.sleep (1) asyncio.run (main ())

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

Diese Asyncresource -Klasse implementiert die

aenter und aexit Methoden, sodass sie mit der Async mit Anweisung verwendet werden kann.

Abschließend bietet das Coroutine -System von Python eine leistungsstarke Grundlage für den Aufbau von benutzerdefinierten asynchronen Primitiven. Durch das Verständnis der zugrunde liegenden Mechanismen und Protokolle können Sie maßgeschneiderte Lösungen für bestimmte asynchrone Herausforderungen erstellen, die Leistung in komplexen gleichzeitigen Szenarien optimieren und die asynchronisierten Funktionen von Python erweitern. Denken Sie daran, dass diese benutzerdefinierten Implementierungen für das Lernen und spezifische Anwendungsfälle hervorragend sind, ist die integrierte Asyncio-Bibliothek von Python sehr optimiert und sollte für die meisten Szenarien Ihre Anlaufstelle sein. Happy Coding!

---

Unsere Kreationen

schauen Sie sich unbedingt unsere Kreationen an:

Investor Central | Smart Living | epochs & echoes | rätselhafte Mysterien | hindutva | elite dev | js Schools

---

Wir sind auf Medium

Tech Koala Insights | epochs & echoes world | Investor Central Medium | rätselhafte Mysterien Medium | Science & Epochen Medium | moderne hindutva