标签: python-asyncio

您好asyncio专家,

如何正确返回协程的结果？我需要使用return或future.set_result吗？

例如:

@coroutine
def do_something()
    result = yield from some_function()
    return result

要么

@coroutine
def do_something()
    future = asyncio.Future()
    result = yield from some_function()
    future.set_result(result)

我编写基于asyncio框架的应用程序。此应用与具有速率限制（每秒最多2次调用）的API进行交互。因此，我将与API交互的方法移到了芹菜中，以将其用作速率限制器。但这看起来像是开销。

有什么方法可以创建新的异步事件循环（或其他方式），以确保协程的执行不超过每秒n次？

我在Windows 32位中使用Python 3.5.2并且知道asyncio call_at不是线程安全的,因此下面的代码不会打印'bomb',除非我取消注释该行 loop._write_to_self().

import asyncio
import threading


def bomb(loop):
    loop.call_later(1, print, 'bomb')
    print('submitted')
    # loop._write_to_self()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    threading.Timer(2, bomb, args=(loop,)).start()
    loop.run_forever()

但是,我无法找到有关原因call_at_threadsafe和call_later_threadsafe实施的信息.这个理由是否存在？

当我在Mac OSX命令行上的python中执行以下操作时:

>>> from concurrent.futures import ProcessPoolExecutor

我明白了

ModuleNotFoundError: No module named 'concurrent.futures'; 'concurrent' is not a package

python --version使用Virtualenvs给人3.6.0,我已经做了pip3 install asyncio,也pip3 install futures.

在Jupyter Notebook(python 3.6.1)中,我在(18.5.3.1.1.示例:Hello World协程)中运行了基本的python docs Hello_World,并注意到它给了我一个RuntimeError.在尝试了很长时间以找到程序问题后(我的理解是文档可能不是完全最新的),我终于注意到它只在第二次运行后执行此操作并在重新启动的内核中进行测试.我从那以后在两个连续的单元格(1和2)中复制了相同的小python程序,发现它在第二个而不是第一个上给出了错误,并在之后给出了错误.这在重新启动内核后重复此操作.

import asyncio

def hello_world(loop):
    print('Hello World')
    loop.stop()

loop = asyncio.get_event_loop()

# Schedule a call to hello_world()
loop.call_soon(hello_world, loop)

# Blocking call interrupted by loop.stop()
loop.run_forever()
loop.close() 

追溯:

RuntimeError                  Traceback (most recent call last)
<ipython-input-2-0930271bd896> in <module>()
  6 loop = asyncio.get_event_loop()
  7 # Blocking call which returns when the hello_world() coroutine
----> 8 loop.run_until_complete(hello_world())
  9 loop.close()

/home/pontiac/anaconda3/lib/python3.6/asyncio/base_events.py in run_until_complete(self, future)
441         Return the Future's result, or raise its exception.
442         """
--> 443         self._check_closed()
444 …

# Example 2: asynchronous requests
import asyncio
import aiohttp
import time
import concurrent.futures
no = int(input("time of per "))
num_requests = int(input("enter the no of threads "))
no_1 = no
avg = 0
async def fetch():
    async with aiohttp.ClientSession() as session:
        await  session.get('http://google.com')

while no > 0:
    start = time.time()
    async def main():
        with concurrent.futures.ThreadPoolExecutor(max_workers=num_requests) as executor:
            loop = asyncio.get_event_loop()
            futures = [
                loop.run_in_executor(
                    executor,
                    fetch
                )
                for i in range(num_requests)
            ]
        for response in await asyncio.gather(*futures):
            pass
    loop = …

是否可以首先为每个任务打开浏览器，然后再打开链接？此代码引发错误

import asyncio
from selenium import webdriver

async def get_html(url):
    driver = await webdriver.Chrome()
    response = await driver.get(url)

TypeError：对象WebDriver不能在“ await”表达式中使用

我似乎很难理解蟒蛇asyncio.我没有编写任何代码,因为我看到的所有示例都是针对一次性运行的.创建一些协同程序,将它们添加到事件循环中,然后运行循环,它们运行在它们之间切换的任务,完成.这似乎对我没什么帮助.

我想使用asyncio不中断我的应用程序中的操作(使用pyqt5).我想创建一些函数,当在asyncio事件循环中调用run时,然后当它们完成时,它们会进行回调.

我的想象是.为asyncio创建一个单独的线程,创建循环并永远运行它.创建一些功能getFile(url, fp),get(url),readFile(file)等于是在UI,我有一个提交按钮的文本框,用户输入网址,点击提交,它下载文件.

但是,我看到的每个例子,我都看不到如何在正在运行的循环中添加协同程序.而且我没有看到如何在不增加运行循环的情况下做我想做的事情.

#!/bin/python3
import asyncio
import aiohttp
import threading

loop = asyncio.get_event_loop()

def async_in_thread(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

async def _get(url, callback):
    print("get: " + url)
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            result = await response.text()
            callback(result)
            return

def get(url, callback):
    asyncio.ensure_future(_get(url, callback))

thread = threading.Thread(target=async_in_thread, args=(loop, ))

thread.start()

def stop():
    loop.close()

def callme(data):
    print(data)
    stop()

get("http://google.com", callme)

thread.join()

这是我的想象,但它不起作用.

免责声明:我是aiohttp的初学者

我正在尝试使用aiohttp异步处理get请求但事实证明它比gevent的池版本慢得多.

GEVENT VERSION

import gevent
from gevent import monkey
monkey.patch_all()
from gevent.pool import Pool

import requests
import time

def pooling_task(url):
    requests.get(url)


def pooling_main():
    start = time.time()
    pool = Pool(10)
    urls = [
        "http://google.com",
        "http://yahoo.com",
        "http://linkedin.com",
        "http://shutterfly.com",
        "http://mypublisher.com",
        "http://facebook.com"
    ]
    for url in urls:
        pool.apply_async(pooling_task, args=(url,))

    pool.join()
    end = time.time()
    print("POOL TIME {}".format(end-start))

if __name__ == '__main__':
    print("POOLING VERSION")
    pooling_main()

输出 - 池时间6.299163818359375

以下是aiohttp版本

import aiohttp
import asyncio
import time
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

async def fetch(session, url):
    async with session.get(url) …

当我意识到同步版本比异步版本快20倍时，我正在将生产系统迁移到异步。我能够创建一个非常简单的示例，以可重复的方式进行演示。

异步版本

import asyncio, time

data = {}

async def process_usage(key):
    data[key] = key

async def main():
    await asyncio.gather(*(process_usage(key) for key in range(0,1000000)))

s = time.perf_counter()
results = asyncio.run(main())
elapsed = time.perf_counter() - s
print(f"Took {elapsed:0.2f} seconds.")

这需要19秒。该代码循环遍历1M个键，data并使用相同的键和值构建一个字典。

$ python3.7 async_test.py
Took 19.08 seconds.

同步版本

import time

data = {}

def process_usage(key):
    data[key] = key

def main():
    for key in range(0,1000000):
        process_usage(key)

s = time.perf_counter()
results = main()
elapsed = time.perf_counter() - s
print(f"Took {elapsed:0.2f} seconds.")

这需要0.17秒！并执行与上述完全相同的操作。

$ …