标签: python-asyncio

我正在使用Asyncio和Requests来对一系列HTTP请求进行基准测试.

出于某种原因,使用Asyncio比使用直接请求稍慢.知道为什么吗？我是否错误地使用Asyncio？

import asyncio
import functools
import requests
import time

ts = time.time()
for i in range(10):
  @asyncio.coroutine
  def do_checks():
      loop = asyncio.get_event_loop()
      req = loop.run_in_executor(None, functools.partial(requests.get, "http://google.com", timeout=3))
      resp = yield from req
      print(resp.status_code)

  loop = asyncio.get_event_loop()
  loop.run_until_complete(do_checks())
te = time.time()
print("Version A: " + str(te - ts))

ts = time.time()
for i in range(10):
  r = requests.get("http://google.com", timeout=3)
  print(r.status_code)
te = time.time()

print("Version B:  " + str(te - ts))

输出:

版本A = Asyncio; 版本B =请求

200
200
200 …

我正在尝试async使用asyncio事件循环创建一种对绑定方法的计时器回调.现在的问题是绑定的异步方法不应该保存对实例的强引用,否则后者永远不会被删除.定时器回调应该只与父实例一样长.我找到了一个解决方案,但我认为它并不漂亮:

import asyncio
import functools
import weakref

class ClassWithTimer:
    def __init__(self):
        asyncio.ensure_future(
            functools.partial(
                ClassWithTimer.update, weakref.ref(self)
            )()
        )

    def __del__(self):
        print("deleted ClassWithTimer!")

    async def update(self):
        while True:
            await asyncio.sleep(1)
            if self() is None: break
            print("IN update of object " + repr(self()))

async def run():
    foo = ClassWithTimer()
    await asyncio.sleep(5)
    del foo

loop = asyncio.get_event_loop()
loop.run_until_complete(run())

有没有更好,更pythonic的方式来做到这一点？计时器回调确实需要异步.没有asyncio,weakref.WeakMethod可能是要走的路.但是asyncio.ensure_future需要一个协程对象,所以在这种情况下它不起作用.

我主要在IPython的交互式shell中工作.有时我会从Internet复制粘贴代码来测试它并使用示例.

如果我粘贴本教程中的以下代码:

import asyncio

async def speak_async():  
    print('OMG asynchronicity!')

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

我将关闭循环.该文件说,要使用在一个事件循环的任何方法,它已经关闭了.并且async.get_event_loop()仍然会返回那个闭环.

如果我意外关闭事件循环该怎么办？我宁愿不重启解释器.

我有一个协程如下:

async def download():
    downloader = DataManager()
    downloader.download()

DataManager.download() 方法看起来像:

def download(self):
    start_multiple_docker_containers()
    while True:
        check_containers_statuses()
        sleep(N)  # synchronous sleep from time module

这是一个好习惯吗？如果没有,我怎么能使用asyncio.sleep的download()？

或许这样的代码结构在概念上是错误的？

我想运行一个使用协同程序和多线程请求URL的服务.但是我无法将协同程序传递给执行程序中的工作程序.有关此问题的最小示例,请参阅下面的代码:

import time
import asyncio
import concurrent.futures

EXECUTOR = concurrent.futures.ThreadPoolExecutor(max_workers=5)

async def async_request(loop):
    await asyncio.sleep(3)

def sync_request(_):
    time.sleep(3)

async def main(loop):
    futures = [loop.run_in_executor(EXECUTOR, async_request,loop) 
               for x in range(10)]

    await asyncio.wait(futures)

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

导致以下错误:

Traceback (most recent call last):
  File "co_test.py", line 17, in <module>
    loop.run_until_complete(main(loop))
  File "/usr/lib/python3.5/asyncio/base_events.py", line 387, in run_until_complete
    return future.result()
  File "/usr/lib/python3.5/asyncio/futures.py", line 274, in result
    raise self._exception
  File "/usr/lib/python3.5/asyncio/tasks.py", line 239, in _step
    result = coro.send(None)
  File "co_test.py", line 10, in main
    futures …

我正在尝试从FTP服务器下载超过30,000个文件,并且在使用异步IO进行一些谷歌搜索之后似乎是一个好主意.但是,下面的代码无法下载任何文件并返回超时错误.我真的很感激任何帮助!谢谢!

class pdb:
    def __init__(self):
        self.ids = []
        self.dl_id = []
        self.err_id = []


    async def download_file(self, session, url):
        try:
            with async_timeout.timeout(10):
                async with session.get(url) as remotefile:
                    if remotefile.status == 200:
                        data = await remotefile.read()
                        return {"error": "", "data": data}
                    else:
                        return {"error": remotefile.status, "data": ""}
        except Exception as e:
            return {"error": e, "data": ""}

    async def unzip(self, session, work_queue):
        while not work_queue.empty():
            queue_url = await work_queue.get()
            print(queue_url)
            data = await self.download_file(session, queue_url)
            id = queue_url[-11:-7]
            ID = id.upper()
            if …

如何定义异步装置并在异步测试中使用它们？

以下代码全部在同一个文件中,失败了.是否由测试跑者明确地称之为夹具并且没有等待？

@pytest.fixture
async def create_x(api_client):
    x_id = await add_x(api_client)
    return api_client, x_id

async def test_app(create_x, auth):
    api_client, x_id = create_x
    resp = await api_client.get(f'my_res/{x_id}', headers=auth)
    assert resp.status == web.HTTPOk.status_code

生产

==================================== ERRORS ====================================
_____________ ERROR at setup of test_app[pyloop] ______________

api_client = <aiohttp.test_utils.TestClient object at 0x7f27ec954f60>

    @pytest.fixture
    async def create_x(api_client):
>       x_id = await add_x(api_client)
...
... cannot show the full trace and pathnames sorry
...    

in __await__
    ret = yield from self._coro /home/mbb/.pyenv/versions/3.6.3/envs/mr/lib/python3.6/site-packages/aiohttp/test_utils.py:245: in request
    method, self.make_url(path), *args, …

作为一项学习练习，我试图修改aiohttp的快速入门示例，以使用单个ClientSession提取多个URL（文档建议通常应为每个应用程序创建一个ClientSession）。

import aiohttp
import asyncio

async def fetch(session, url):
  async with session.get(url) as response:
    return await response.text()

async def main(url, session):
  print(f"Starting '{url}'")
  html = await fetch(session, url)
  print(f"'{url}' done")

urls = (
  "https://python.org",
  "https://twitter.com",
  "https://tumblr.com",
  "https://example.com",
  "https://github.com",
)

loop = asyncio.get_event_loop()
session = aiohttp.ClientSession()
loop.run_until_complete(asyncio.gather(
  *(loop.create_task(main(url, session)) for url in urls)
))
# session.close()   <- this doesn't make a difference

但是，在协程之外创建ClientSession显然不是可行的方法：

？python 1_async.py
1_async.py:30：用户警告：在协程之外创建客户端会话是一个非常危险的想法
  会话= aiohttp.ClientSession（）
在协程之外创建客户端会话
client_session： 
开始'https://python.org'
开始“ https://twitter.com”
开始'https://tumblr.com'
开始“ https://example.com”
开始'https://github.com'
'https://twitter.com'完成
“ https://example.com”完成
'https://github.com'完成 …

我正在尝试创建两个线程,每个线程都有自己的asyncio事件循环.

我尝试了以下代码,但它似乎不起作用:

import asyncio
from threading import Thread

def hello(thread_name):
    print('hello from thread {}!'.format(thread_name))

event_loop_a = asyncio.new_event_loop()
event_loop_b = asyncio.new_event_loop()

def callback_a():
    asyncio.set_event_loop(event_loop_a)
    asyncio.get_event_loop().call_soon_threadsafe(lambda: hello('a'))

def callback_b():
    asyncio.set_event_loop(event_loop_b)
    asyncio.get_event_loop().call_soon_threadsafe(lambda: hello('b'))

thread_a = Thread(target=callback_a, daemon=True)
thread_b = Thread(target=callback_b, daemon=True)
thread_a.start()
thread_b.start()

我的用例是调用Tornado web框架的websocket_connect异步函数.

我有一个异步应用程序,它通过aiohttp提供请求并执行其他异步任务(与数据库交互,处理消息,将请求本身作为HTTP客户端).我想监视事件循环的繁忙程度,也就是说它花费了多少时间来执行代码而不是等待select完成.

有没有办法用标准库事件循环或其他第三方循环(uvloop)来衡量这个？

具体来说,我想要一个持续的百分比衡量饱和度,而不仅仅是这个问题似乎要解决的二元"忙碌" .