标签: python-asyncio

我正在使用Python 3 asyncio框架评估不同的模式以定期执行(实际的睡眠/延迟以简化),我有两段代码表现得很差,我无法解释原因.yield from正如我所料,第一个版本用于递归调用自身耗尽大约1000次迭代的堆栈.第二个版本递归调用协程,但委托实际的事件循环执行到asyncio.async并且不会耗尽堆栈.你能详细解释为什么第二个版本没有使用堆栈吗？执行此协程的两种方式有什么区别？

第一版(收益率):

@asyncio.coroutine
def call_self(self, i):
    print('calling self', i)
    yield from self.call_self(i + 1)

第二个版本(asyncio.async):

@asyncio.coroutine
def call_self(self, i):
    print('calling self', i)
    asyncio.async(self.call_self(i + 1))

一个python的新手,花了很多时间阅读文档和其他代码我似乎无法asyncio在Python 3中获得新模块.它一直没有堆栈跟踪终止给我一个线索,应该永远运行但不会.

我试图模仿的基本流程概念如下:

从端口读取: 打开端口 - >读取数据(可变长度) - >放在queue1上

然后处理数据: 从queue1获取数据 - >条件适用 - >结果放在queue2上

然后写入端口: 从queue2获取数据并写入端口

永远地从顶部循环

注意:输入端口上的数据是偶发的,可变长度,因此我可以使用几个块来"序列" asyncio.我理解asyncio将允许一个块到达的情况,然后在我的应用程序响应之前的另一个 - 即该调用get_io_from_port()有助于协同例程的多次执行.这就是我使用队列来确保不阻塞的原因process_queue()

到目前为止我的玩具示例代码:

import queue
import asyncio

@asyncio.coroutine
def process_queue(q1, q2):
    tmp = q1.Get()
    if tmp == 'ABCDEF':
        q2.put('12345')
    elif tmp == 'GHIJKL':
        q2.put =('67890')
    else:
        print('There is a data error')

@asyncio.coroutine
def put_io_to_port(writer, q2):

    if not q2.empty():
        try:
                writer.write(q2.get())

        except IOError as e:
            print('OUT Port issue: ', e)

@asyncio.coroutine
def get_io_from_port(reader, …

使用旧学校subprocess.Popen()API 启动两个进程时,我可以轻松地将标准从一个进程连接到另一个进程的标准进程,创建一个管道,就像UNIX shell在连接命令时所做的那样|:

from subprocess import Popen, PIPE

process_1 = Popen(['ls'], stdout = PIPE)
process_2 = Popen(['wc'], stdin = process_1.stdout)

process_1.wait()
process_2.wait()

在使用asyncio.subprocess.create_subprocess_exec()(或类似的)异步API时,如何实现相同的目标？这是我试过的:

from asyncio.events import get_event_loop
from asyncio.subprocess import PIPE, create_subprocess_exec

async def main():
    process_1 = await create_subprocess_exec('ls', stdout = PIPE)
    process_2 = await create_subprocess_exec('wc', stdin = process_1.stdout)

    await process_1.wait()
    await process_2.wait()

get_event_loop().run_until_complete(main())

但第二次打电话create_subprocess_exec()抱怨传递的论点stdin没有fileno(这是真的):

Traceback (most recent call last):
  File ".../test-async.py", line 11, …

当我做出一个奇怪的发现时,我决定使用Python 实现睡眠排序(https://rosettacode.org/wiki/Sorting_algorithms/Sleep_sort)asyncio:它使用负值(并立即返回0)!

这是代码(你可以在这里运行https://repl.it/DYTZ):

import asyncio
import random

async def sleepy(value):
    return await asyncio.sleep(value, result=value)


async def main(input_values):
    result = []
    for sleeper in asyncio.as_completed(map(sleepy, input_values)):
        result.append(await sleeper)
    print(result)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    input_values = list(range(-5, 6))
    random.shuffle(input_values)
    loop.run_until_complete(main(input_values))

正如预期的那样,代码需要5秒才能执行,但结果总是如此[0, -5, -4, -3, -2, -1, 1, 2, 3, 4, 5].我可以理解0立即返回,但负值如何以正确的顺序返回？

我正在阅读包中的文档Protocol和Transport类asyncio.特别:

子类化协议类时,建议您覆盖某些方法.这些方法是回调:它们将由传输在某些事件上调用(例如,当收到某些数据时); 你不应该自己打电话,除非你正在实施运输.

强调补充说

因此,原则上应该可以实现运输,但......

传输是asyncio提供的类,用于抽象各种通信信道.您通常不会自己实例化运输; 相反,您将调用一个AbstractEventLoop方法 (哪一个？),它将创建传输并尝试启动底层通信通道,并在成功时回拨您.

再次强调了重点

阅读关于AbstractEventLoop的部分,我没有看到任何创建自定义传输的方法.它最接近的AbstractEventLoop.create_connection(protocol_factory, host=...)只是暗示它会创造某种插座......

好吧,我的最终目标是使用一个自定义传输,它不是任何类型的套接字(也许StringIO是一个数据库连接游标,也许是另一种协议).

那么,这只是一个很好的但是从未在文档中实现过的错误,或者实际上是否有一种方法可以实现自定义传输而无需修补猴子asyncio并牺牲第一胎？

我有以下代码

import asyncio

loop = asyncio.get_event_loop()

async def while_loop():
    n = 0
    while True:
        print(f"{n}")
        await asyncio.sleep(2)
        n = n+1

async def some_func():
    await asyncio.sleep(5)
    print("Some Func")

future = loop.create_task(while_loop())
loop.run_until_complete(some_func())

我希望while_loop函数永远运行,但它似乎只是作为调用的结果执行,run_until_complete并且一旦some_func完成执行就停止打印while循环.我看到的输出是:

0
1
2
Some Func

我预计数字会在some_func完成后继续打印.

0
1
2
Some Func
3
4
5
6
.
.
.

获得更多数字的唯一方法是some_func再次打电话.

我是一个新手python3爱好者,他希望了解如何通过库实现asyncio.sleep().

我能够编写协同程序,但我似乎无法考虑如何开始编写异步睡眠代码.

如果您直接共享代码或提供有关如何在本地计算机上查找代码的快速教程,那将会很棒.谢谢!

我正在尝试将消息发送到服务器以获取答案。

我曾尝试使用该站点上的官方websocket API，但我不理解它们或无法使其按预期运行，因此我正在尝试构建它。

import asyncio
import websockets

 async def test():

     async with websockets.connect('wss://www.bitmex.com/realtime') as websocket:

        await websocket.send("ping")
  #OR   await websocket.send({"op": "subscribe", "args": [<SubscriptionTopic>]})

        response = await websocket.recv()
        print(response)

 asyncio.get_event_loop().run_until_complete(test())

我收到了连接的消息，但我没有收到“ pong”作为对“ ping”的回答，也没有收到“好，您已订阅此主题”，因为在回显网站上尝试执行命令时，我没有收到。

#!/usr/bin/env python3

import asyncio
import websockets
import json

var = []

async def test():
async with websockets.connect('wss://www.bitmex.com/realtime') as websocket:
    response = await websocket.recv()
    print(response)


    await websocket.send(json.dumps({"op": "subscribe", "args": "trade:TRXH19"}))
    response = await websocket.recv()

    resp = await websocket.recv()
    print(json.loads(resp))

    sum=0

    while True:

        resp = await websocket.recv()
        jj = json.loads(resp)["data"][0] …

我如何设置阻塞函数在执行程序中运行，其结果无关紧要，因此主线程不应等待或被其拖慢。

老实说，我不确定这是否是正确的解决方案，我只想将某种类型的处理队列与主进程分开，以免它阻止服务器应用程序返回请求，因此Web服务器类型运行一个工作线程处理许多请求。

最好是我想远离像Celery这样的解决方案，但是如果那是最理想的，我将愿意学习。

这里的上下文是一个异步Web服务器，该服务器生成具有大图像的pdf文件。

app = Sanic()
#App "global" worker
executor = ProcessPoolExecutor(max_workers=5)

app.route('/')
async def getPdf(request):
  asyncio.create_task(renderPdfsInExecutor(request.json))
  #This should be returned "instantly" regardless of pdf generation time
  return response.text('Pdf being generated, it will be sent to your email when done')

async def renderPdfsInExecutor(json):
  asyncio.get_running_loop.run_in_executor(executor, syncRenderPdfs, json)

def syncRenderPdfs(json)
  #Some PDF Library that downloads images synchronously
  pdfs = somePdfLibrary.generatePdfsFromJson(json)
  sendToDefaultMail(pdfs)

上面的代码给出了错误（是的，它以admin身份运行）：

PermissionError [WinError 5] Access denied
Future exception was never retrieved

奖励问题：通过在执行程序内部运行asyncio循环，我可以获得任何收益吗？因此，如果它一次处理多个PDF请求，它将在它们之间分配处理。如果是，该怎么办？

我正在尝试为websockets客户端学习异步。我尝试的每一段代码都会出现以下错误：

RuntimeError：无法从正在运行的事件循环中调用asyncio.run（）

我尝试了最简单的代码，它始终会给出RuntimeError。我尝试再次安装完整的anaconda发行版，等等，但找不到问题所在。

我正在将Spyder 3.3.3与Python 3.7.3结合使用

可以正常工作的代码示例：

import asyncio

async def main():
    print('hello')
    await asyncio.sleep(1)
    print('world')

asyncio.run(main())

错误信息：

File "C:\Users\jmart\Anaconda3\lib\site-packages\spyder_kernels\customize\spydercustomize.py", line 786, in runfile
  execfile(filename, namespace)
File "C:\Users\jmart\Anaconda3\lib\site-packages\spyder_kernels\customize\spydercustomize.py", line 110, in execfile
  exec(compile(f.read(), filename, 'exec'), namespace)
File "C:/Users/jmart/Documents/asynk2.py", line 8, in <module>
  asyncio.run(main())
File "C:\Users\jmart\Anaconda3\lib\asyncio\runners.py", line 34, in run
  "asyncio.run() cannot be called from a running event loop")
RuntimeError: asyncio.run() cannot be called from a running event loop