Python异步:同步对共享对象的所有访问

Question

我有一个类，它使用asyncio异步处理大量工作元素(主要是由于重叠的HTTP连接请求)。一个非常简单的示例演示了我的代码的结构：

class Work:
    ...

    def worker(self, item):
        # do some work on item...
        return

    def queue(self):
        # generate the work items...
        yield from range(100)

    async def run(self):
        with ThreadPoolExecutor(max_workers=10) as executor:
            loop = asyncio.get_event_loop()
            tasks = [
                loop.run_in_executor(executor, self.worker, item)
                for item in self.queue()
            ]
            for result in await asyncio.gather(*tasks):
                pass

work = Work()
asyncio.run(work.run())

实际上，工作人员需要访问一个类似容器的共享对象，并调用它的方法，这些方法不是async-safe。例如，假设worker方法调用一个定义如下的函数：

def func(shared_obj, value):
    for node in shared_obj.filter(value):
        shared_obj.remove(node)

但是，从工作人员调用func可能会影响此或涉及共享对象的任何其他函数中的其他异步工作人员。我知道我需要使用一些同步，例如全局锁，但是我发现它的使用并不容易：

• asyncio.Lock只能在async函数中使用，因此我必须将所有此类函数定义标记为async。
• 我还必须对这些函数的所有调用进行await
• await也只能在async函数中使用，因此最终worker和func之间的所有函数都将是async。
• 如果worker是async，则不可能将其传递给loop.run_in_executor (它不传递await)。

此外，我必须添加async的一些函数可能是通用的，因为它们应该可以从异步和“正常”上下文中调用。

我可能在整个概念中遗漏了一些很严重的东西。使用threading模块，我只需创建一个锁并在几个地方使用它，而不必对函数进行进一步注释。另外，包装共享对象有一个很好的解决方案，使得所有访问都透明地由一个锁来保护。我想知道asyncio是否有类似的东西.

Answer 1

我可能在整个概念中遗漏了一些很严重的东西。有了线程模块，我就会创建一个锁.

您所缺少的是您根本没有真正使用asyncio。run_in_executor用于将CPU绑定或遗留同步代码集成到异步应用程序中。它的工作方式是将函数提交给ThreadPoolExecutor，并返回一个可处理的句柄，该句柄一旦函数完成就会得到解析。这是在后台运行的“异步”，但不是异步的核心。异步程序由非阻塞部分组成，在数据不可用时使用异步/等待来挂起执行，并依赖事件循环一次有效地等待多个事件并恢复适当的异步函数。

换句话说，只要您依赖run_in_executor，您就只是在使用threading (更确切地说，是带有线程执行器的concurrent.futures )。您可以使用threading.Lock在函数之间进行同步，事情将完全像在一开始使用threading一样工作。

要获得异步的好处，例如扩展到大量并发任务或可靠的取消，您应该从底层将程序设计为异步(或主要是异步)。然后，您可以简单地在两个等待之间修改共享数据，或者使用asyncio.Lock进行跨等待的同步修改。