问题标签 [python-asyncio]

我承认我很懒惰：我需要快速完成这项工作，并且无法理解 Python3 asyncio 模块。（有趣的是，我发现 boost 相当直观。）

我需要读取一个不时阻塞的文件对象（管道）。在此期间，我希望能够以设定的时间间隔（比如每 30 分钟）触发另一个活动，而不管是否可以从文件中读取任何内容。

任何人都可以使用 python3 asyncio 帮助我使用骨架来执行此操作吗？（我无法安装第三方模块，例如twisted。）

I need to run a background job in my pyramid application, one time every hour. The job work asynch without block the main thread that serve responses to the clients requestes. This job need to access to a database for store information.

How can i do this? What library/package/algoritm I have to use?

我在 Python 3.4 中使用 Asyncio，我将尝试解释我目前正在做什么以及我（认为）是什么导致了这个问题。

一方面，我有一个带有阻塞操作的 UDP 连接框架，我从这个流中获取数据并创建我以 SSE 格式传递给客户端的 json。这一切都很好。

我遇到的问题是，如果我不做任何事情并且客户端断开连接，我将无法正确处理客户端断开连接，我将开始收到此错误：

由于循环仍在运行，我一直在寻找彻底打破循环并触发 .close() 的方法，但我在找到的示例中遇到了问题，并且在线资源并不多。

似乎实际可行的一个示例是尝试从客户端读取一行，如果它是一个空字符串，则意味着客户端已断开连接。

但是在大约十条消息之后，发送给客户端的所有消息都停止了，我认为这是因为它挂在“data =（来自client_reader.readline（））”上，如果我关闭客户端然后它会正确关闭并“结束”连接”确实被调用。任何想法为什么它可能会挂起？我认为我在这一点上对 Asyncio 有一个很好的处理，但这个让我感到困惑。

注意：location() 和 status() 是我从 UDP 套接字获取信息的两个调用——我已经使用相同的代码成功运行了它们多个小时而没有出现问题——减去客户端断开线路。

更新：如果我在“来自 client_reader 的收益”上添加超时，当它达到它通常会挂起的程度时，我会收到以下错误。

这是一个示例脚本，显示了运行中的错误 - 只需在 python 3.4.2 中运行它，经过 9 次迭代后，它将挂起从客户端读取。

（脚本已完成，您可以运行它自己查看）

另一个更新：我发现它死了，因为它从客户端的标题中读取所有行，然后在用完行时超时。我认为，我正在寻找的真正答案是，当您实际上不需要从客户端接收数据（除了初始连接）时，如何检测客户端断开连接。

我在 Debian 7 服务器上的 Python 3.4.2 下遇到了一个奇怪的 Asyncio 问题。我正在尝试从客户端读取以检测它们是否断开连接，网络上的大多数示例都使用此方法，如果您读取响应并且它是空的，则您会中断循环并关闭连接。

然而，恰好在 9 次迭代之后，脚本将在读取行中终止，我什至尝试只在每 10 次迭代时进行读取，然后它会在 90 次迭代后终止。如果我不尝试检测客户端断开连接，脚本将愉快地永远运行而不会出现问题。

原始代码基于此要点的服务器： https ://gist.github.com/dbehnke/9627160

有什么想法可能导致这种情况吗？

（脚本已完成，您可以运行它自己查看）

在 Python 3.4/Asyncio 中，我使用的是 StreamReader/Writer。

要检测客户端断开连接，常用方法似乎是从客户端读取，如果那里什么都没有，则客户端断开连接。

但是，您很快就用完了从客户端标头中读取的行（它移动到每个循环的下一行），并且如果没有从客户端发送的其他行（在我的情况下，客户端不只是发送侦听），您点击暂停。

我想做的是只一遍又一遍地读取标题的第一行..或者甚至可能只是第一行的第一个字符，或者当它到达最后一行时不可能循环回到第一的。

完成这项任务的最佳/最优雅的方式是什么？3.4/Asyncio/StreamReader/Writer。（检测客户端断开连接）

Guido van Rossum 在 2014 年关于 Tulip/Asyncio 的演讲中展示了这张幻灯片：

任务与协程

• 相比：

  • res = 来自 some_coroutine(...)
  • res = 来自任务的产量（some_coroutine（...））

• 任务无需等待即可取得进展

  • 就像你在等待别的东西一样记录
    • 即产量来自

我完全没有抓住重点。

从我的角度来看，这两种结构是相同的：

在裸协程的情况下 - 它被调度，因此无论如何都会创建任务，因为调度程序与任务一起运行，然后协程调用者协程被挂起，直到被调用者完成，然后可以自由继续执行。

万一Task- 都一样 - 新任务被调度并且调用者协程等待它的完成。

这两种情况下代码执行的方式有什么不同，开发人员在实践中应该考虑什么影响？

ps
非常感谢权威来源（GvR、PEP、文档、核心开发人员说明）的链接。

通过zmq.Socket.send_multipart以非阻塞方式发送数据时，如何判断操作是否失败？例如，在aiozmq library中，这是我要调试的：

我的场景是服务器通过 zmq.Socket.send_multipart 发送数据，但客户端从未收到它，所以我想知道数据的发送是否只是失败了。

https://docs.python.org/3/library/asyncio-eventloop.html#delayed-calls

注意：超时（相对延迟或绝对时间）不应超过一天。

有人对为什么存在这个限制有很好的解释吗？

提前致谢

我正在编写简单的生产者/消费者程序。

消费者似乎没有机会执行。

套接字每 500 毫秒接收一次数据，因此当yield from在 receive_data 函数中挂起生产者协程时，消费者协程将打印信息。

这有什么可以解释的？

考虑这段代码：

这是由生成器构建的管道的图示。我在实践中经常使用这种模式来构建数据处理流程。这就像 UNIX 管道。

将生成器重构为每次暂停执行的协程的最优雅方法是什么yield？

更新

我的第一次尝试是这样的：

但由于某种原因它不起作用 - 变量a变为None.

更新#1

我最近想出的：

服务器：

客户：

与“David Beazley 的协程”相比的优势在于，您可以将asyncio此类处理单元内的所有内容input与output队列一起使用。
这里的缺点 - 连接管道单元需要很多队列实例。它可以通过使用比asyncio.Queue.
另一个缺点是这种处理单元不会将它们的异常传播到父堆栈帧，因为它们是“后台任务”，而“David Beazley 的协程”会传播。

更新#2

这就是我想出的：
https ://gist.github.com/AndrewPashkin/04c287def6d165fc2832