如何在 Python 3 中的多个异步方法之间创建依赖关系？

Question

我有一个包含多个异步方法的类，我想在它们之间创建依赖关系。 Dependency Graph 在这张图中，heartbeat 和neighbor_check 都依赖于radio，sub 依赖于neighbor_check，这意味着在启动radio 之后我想启动heartbeat 和neighbor_check。 在开始neighbour_check 之后，我想启动sub。我还有一个 asyinc_start 函数来启动这些方法，所以我想在这里管理这些方法，但我不能。

给我一些建议。

check_dependency.py的代码如下：

import asyncio
import sys
import os
from util import run`enter code here`
class ChkDependency():
    async def radio(self):
        print("Radio is initialized")
        await asyncio.sleep(2)


    async def pub(self):
        await asyncio.sleep(2)
        print("pub is initialized")


    async def heartbeat(self):
        print("heartbeat started")
        await asyncio.sleep(2)


    async def neigh_hood_check(self):
        await asyncio.sleep(2)
        print("checking for matches in neigh list")


    async def subs(self):
        await asyncio.sleep(2)
        print("match found")
        await asyncio.sleep(2)
        print("subscribing.....")


if __name__ == '__main__':
    chkdependency = ChkDependency()
    async def start():
        while True:
            await asyncio.wait([
                chkdependency.radio(),
                chkdependency.pub(),
                chkdependency.heartbeat(),
                chkdependency.neigh_hood_check(),
                chkdependency.subs(),
            ])    

    try:
        run(
            start(),
          )
    except KeyboardInterrupt:
        print("Exiting...")
        exit()

---

util.py的代码如下：

import asyncio

def run(*args):
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.gather(*args))

我正在考虑使用信号量来实现依赖，但没有积极的结果。 请帮忙！！

Answer 1

从问题的描述来看，不清楚你想知道什么，所以我假设一下。

1. 如果你只想让方法按照给定的顺序执行，那么 这样做：

   try:
       while True:
           await chkdependency.radio()
           await chkdependency.pub()
           await chkdependency.heartbeat()
           await chkdependency.neigh_hood_check()
           await chkdependency.subs()
   

   但是在这种情况下，每个后续方法都会等待 完成上一个！

2. 如果您希望方法保证在指定的 顺序，并以任何顺序结束，那么同步原语是 不可或缺：

   import asyncio
   from random import randint
   
   
   class ChkDependency:
       def __init__(self):
           self.radio_started = asyncio.Event()
           self.pub_started = asyncio.Event()
           self.heartbeat_started = asyncio.Event()
           self.neigh_hood_check_started = asyncio.Event()
           self.sub_started = asyncio.Event()
   
       async def radio(self):
           await self.radio_started.wait()
           print("1 Start radio")
           self.pub_started.set()
           await asyncio.sleep(randint(1, 5))
           print("1 Finish radio")
           self.radio_started.clear()
   
       async def pub(self):
           await self.pub_started.wait()
           print("2 Start pub")
           self.heartbeat_started.set()
           await asyncio.sleep(randint(1, 5))
           print("2 Finish pub")
           self.pub_started.clear()
   
       async def heartbeat(self):
           await self.heartbeat_started.wait()
           print("3 Start heartbeat")
           self.neigh_hood_check_started.set()
           await asyncio.sleep(randint(1, 5))
           print("3 Finish heartbeat")
           self.heartbeat_started.clear()
   
       async def neigh_hood_check(self):
           await self.neigh_hood_check_started.wait()
           print("4 Start neigh_hood_check")
           self.sub_started.set()
           await asyncio.sleep(randint(1, 5))
           print("4 Finish neigh_hood_check")
           self.neigh_hood_check_started.clear()
   
       async def subs(self):
           await self.sub_started.wait()
           print("5 Start subs")
           await asyncio.sleep(randint(1, 5))
           print("5 Finish subs")
           self.sub_started.clear()
   
       async def start(self):
           while True:
               self.radio_started.set()
               print("\nStart new cycle:")
               await asyncio.gather(
                   self.radio(),
                   self.pub(),
                   self.heartbeat(),
                   self.neigh_hood_check(),
                   self.subs()
               )
   
   
   async def main():
       chkdependency = ChkDependency()
       await chkdependency.start()
   
   if __name__ == '__main__':
       try:
           asyncio.run(main())
       except KeyboardInterrupt:
           print("Exiting...")
   

3. 请记住，如果您这样做（或与 asyncio.wait 相同） 而不是 asyncio.gather):

   async def start():
       while True:
           await asyncio.gather(
               chkdependency.radio(),
               chkdependency.pub(),
               chkdependency.heartbeat(),
               chkdependency.neigh_hood_check(),
               chkdependency.subs()
           )
   

   在您看来，这些方法似乎是依次开始的。那是因为 在内部， asyncio.gather 和 asyncio.wait 都在循环中运行它们。 但是，请记住，不能保证这种行为！和这个 意味着在任何时候，例如，在新版本的 库或出于某些内部原因，这些方法可以从 顺序不同。

4. 您似乎也不需要 run 方法，因为 asyncio.run 需要 一样！