标签: python-asyncio

这是取消任务的示例：

import asyncio


async def some_func():
    await asyncio.sleep(2)
    print('Haha! Task keeps running!')
    await asyncio.sleep(2)

async def cancel(task):
    await asyncio.sleep(1)
    task.cancel()

async def main():
    func_task = asyncio.ensure_future(some_func())
    cancel_task = asyncio.ensure_future(cancel(func_task))
    try:
        await func_task
    except asyncio.CancelledError:
        print('Task cancelled as expected')

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

# Task cancelled as expected
# [Finished in 1.2s]

它工作正常，任务被取消。如果CancelledError被捕获到内部some_func任务不会被取消：

async def some_func():
    try:
        await asyncio.sleep(2)
    except:
        pass
    print('Haha! Task keeps running!')
    await asyncio.sleep(2)

# Haha! Task keeps running!
# …

我使用aioredis编写了异步服务，该服务将在某个频道上侦听并以异步方式运行一些命令。

基本上，我从示例页面中获取了一个代码来编写一个小型测试应用程序，并删除了不必要的部分：

import asyncio
import aioredis

async def reader(ch):
    while (await ch.wait_message()):
        msg = await ch.get_json()
        print('Got Message:', msg)
        i = int(msg['sleep_for'])
        print('Sleep for {}'.format(i))
        await asyncio.sleep(i)
        print('End sleep')


async def main():
    sub = await aioredis.create_redis(('localhost', 6379))
    res = await sub.subscribe('chan:1')
    ch1 = res[0]
    tsk = await reader(ch1)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

还有另一个测试应用程序，它发布带有sleep_for字段的json blob ，然后在订阅者应用程序中使用该字段reader使用sleep语句来模拟协程内部的某些工作。

我希望“睡眠”以“并行”方式运行，但实际上它们以同步的方式出现在屏幕上，一个接一个。

我的猜测是，只要打到await ch.get_json(..)（或什至await ch.wait_message()）行，我就应该能够处理下一条消息。在实践中，它像同步代码一样运行。我哪里错了？这可以使用连接池来处理，但这意味着存在一些不异步的问题，也不知道确切的含义。

我有一个asyncio TCP服务器从客户端接收消息,在服务器上执行stuff()并发回文本.服务器在正确接收和发送数据的意义上运行良好.问题是我无法从客户端的服务器取回消息,因为我在控制台输入时有阻塞例程(基本上从不执行data_received方法).只有exit命令工作正常(它关闭循环).怎么解决这个？这是服务器和客户端代码.它基本上是EchoClient asyncio版本,还有一些练习的管道代码.

# client.py
import abc
import asyncio
import sys

MENU = '''
a) do x
b) do y
c) exit
'''

loop_ = asyncio.get_event_loop()


class XCommand:
    def run(self):
        self.client.send_data_to_tcp('X:')  # to bytes


class YCommand(Command):
    def run(self):
         s = input('Input for Y ###  ')
         self.client.send_data_to_tcp('Y:' + s)


class ExitCommand(Command):
    def run(self):
        self.client.send_data_to_tcp('EXIT:')
        print('Goodbye!')
        loop_.close()
        exit()


class CommandFactory:
    _cmds = {'a': ACommand,
         'b': BCommand,
         'c': ExitCommand,
         }

    @classmethod
    def get_cmd(cls, cmd):
        cmd_cls = cls._cmds.get(cmd)
        return cmd_cls


def show_menu(client):
    print(MENU)
    while …

是否可以在python3.6中提取协程对象的函数和参数？

上下文：目前我有这样的事情：

async def func(*args):
    ...

ret = await remotely(func, x, y)

在幕后，remotelypickles func,x和y, scp 的那个到不同的服务器，在那里它解压它们，执行func(x,y)，pickles 结果， scp 的那个，最后将它解压到ret.

这个 API 对我来说很讨厌，我更喜欢：

ret = await remotely(func(x, y))

如果我可以腌制由 表示的协程对象func(x, y)，我就可以做到这一点，但是当我尝试这样做时，我得到：

TypeError: can't pickle coroutine objects

所以我的备用的希望是，我可以提取f，x以及y从f(x, y)，因此这个问题。

使用gevent，我可以像这样监视长时间运行的功能：

    greenlet = gevent.Greenlet(long_running_task, *args, **kwargs)
    greenlet.start()
    while not greenlet.ready():
        send_heartbeat()
        gevent.sleep(heartbeat_interval)
    if greenlet.successful():
        send_success(greenlet.value)
    else:
        send_failure(exception=greenlet.exception)

我该如何使用asyncio？我尝试了以下方法，但遇到了麻烦：

    loop = asyncio.get_event_loop()
    async def send_heartbeat(heartbeat_interval=15):
        send_heartbeat()
        asyncio.sleep(heartbeat_interval)

    await asyncio.sleep(1)
    loop.run_until_complete(asyncio.wait([long_running_task(*args, **kwargs), send_heartbeat()]))
    loop.close()

我将如何改变它以获得预期的行为？

我正在尝试使用 asyncio 来加快调用外部工具来分析多个音频文件的过程。我正在 Windows、python 3.6（anaconda 安装）上工作，我在这里遇到的问题是调用似乎没有等待，或者从未通过 stdout 收到结果。

有任何想法吗？

import os
import asyncio

external_tool = r"C:\path\to\external\tool.exe"

def filepaths_generator(root):
    '''
    Simple generator to yield filepaths
    '''
    for path, dirs, files in os.walk(root):
        for f in files:
            yield os.path.join(path,f)


async def async_subprocess_command(*args):
    '''
    A function to run the asyncio subprocess call
    '''
    # Create subprocess
    process = asyncio.create_subprocess_exec(
        *args,
        # stdout must a pipe to be accessible as process.stdout
        stdout=asyncio.subprocess.PIPE)
    # Wait for the subprocess to finish
    stdout, stderr = await process.communicate() …

我是这个框架的新手，正在尝试了解使用一个框架的区别/好处

我可以使用一个非协程的非异步函数

def while_naughty():
    print("naughty")

loop.call_soon(self.while_naughty)

或者

async def naughty():
    print("naughty")

task = asyncio.ensure_future(naughty())

我知道ensure_future以一个协程作为参数，希望了解一下使用call_soon()over更有利的情况ensure_future()。

我正在使用 Python 的 asyncio 模块和async/await并发处理块中的字符序列并将结果收集在列表中。为此，我使用了块功能 ( split) 和块处理功能 ( process_chunk)。它们都来自第三方库，我不想更改它们。

分块很慢，并且事先不知道块的数量，这就是为什么我不想一次消耗整个块生成器。理想情况下，代码应该与process_chunk的信号量同步推进生成器，即每次函数返回时。

我的代码

import asyncio

def split(sequence):
    for x in sequence:
        print('Getting the next chunk:', x)
        yield x
    print('Finished chunking')

async def process_chunk(chunk, *, semaphore=asyncio.Semaphore(2)):
    async with semaphore:
        print('Processing chunk:', chunk)
        await asyncio.sleep(3)
        return 'OK'

async def process_in_chunks(sequence):
    gen = split(sequence)
    coro = [process_chunk(chunk) for chunk in gen]
    results = await asyncio.gather(*coro)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(process_in_chunks('ABC'))

作品和版画种类

import asyncio …

我仍然使用异步python的基础知识，有些事情使我感到困惑。

import asyncio
loop=asyncio.get_event_loop()
for variation in args:
    loop.create_task(coroutine(variation))
loop.run_forever()

似乎与此非常相似

import asyncio
loop=asyncio.get_event_loop()
loop.run_forever(
    asyncio.gather(
        coroutine(variation_1),
        coroutine(variation_2),
        ...))

他们可能会做同样的事情，但这似乎没有用，所以有什么区别？

我正在尝试学习在Python中使用asyncio优化脚本。我的示例返回coroutine was never awaited警告，您可以帮助您理解并找到解决方法吗？

import time 
import datetime
import random
import asyncio

import aiohttp
import requests

def requete_bloquante(num):
    print(f'Get {num}')
    uid = requests.get("https://httpbin.org/uuid").json()['uuid']
    print(f"Res {num}: {uid}")

def faire_toutes_les_requetes():
    for x in range(10):
        requete_bloquante(x)

print("Bloquant : ")
start = datetime.datetime.now()
faire_toutes_les_requetes()
exec_time = (datetime.datetime.now() - start).seconds
print(f"Pour faire 10 requêtes, ça prend {exec_time}s\n")

async def requete_sans_bloquer(num, session):
    print(f'Get {num}')
    async with session.get("https://httpbin.org/uuid") as response:
        uid = (await response.json()['uuid'])
    print(f"Res {num}: {uid}")

async def faire_toutes_les_requetes_sans_bloquer():
    loop = asyncio.get_event_loop()
    with aiohttp.ClientSession() …