标签: python-asyncio

我正在尝试编写一个服务器端事件服务器,我可以使用telnet连接它,并将telnet内容推送到浏览器.使用Python和asyncio背后的想法是使用尽可能少的CPU,因为它将在Raspberry Pi上运行.

到目前为止,我有以下使用库中的库:https://pypi.python.org/pypi/asyncio-sse/0.1,它使用asyncio.

我还复制了一个使用asyncio的telnet服务器.

两者都分开工作,但我不知道如何将两者结合在一起.据我了解,我需要调用send()在SSEHandler从内部类Telnet.data_received,但我不知道如何访问它.这两个"服务器"都需要在循环中运行才能接受新连接或推送数据.

任何人都可以帮忙,或指向另一个方向？

import asyncio
import sse

# Get an instance of the asyncio event loop
loop = asyncio.get_event_loop()

# Setup SSE address and port
sse_host, sse_port = '192.168.2.25', 8888

class Telnet(asyncio.Protocol):
    def connection_made(self, transport):
        print("Connection received!");
        self.transport = transport

    def data_received(self, data):
        print(data)
        self.transport.write(b'echo:')
        self.transport.write(data)

        # This is where I want to send data via SSE
        # SSEHandler.send(data)

        # Things I've tried :(
        #loop.call_soon_threadsafe(SSEHandler.handle_request());
        #loop.call_soon_threadsafe(sse_server.send("PAH!"));

    def connection_lost(self, esc): …

我有5,00,000个网址.并希望异步获得每个响应.

import aiohttp
import asyncio    

@asyncio.coroutine
def worker(url):
    response = yield from aiohttp.request('GET', url, connector=aiohttp.TCPConnector(share_cookies=True, verify_ssl=False))
    body = yield from response.read_and_close()

    print(url)

def main():
    url_list = [] # lacs of urls, extracting from a file

    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait([worker(u) for u in url_list]))

main()

我一次想要200个连接(并发200个),而不是因为这个

当我运行这个程序50个网址它工作正常,url_list[:50] 但但如果我通过整个列表,我得到这个错误

aiohttp.errors.ClientOSError: Cannot connect to host www.example.com:443 ssl:True Future/Task exception was never retrieved future: Task()

可能是频率太高,服务器拒绝在限制后做出响应？

我正在尝试利用Python的新asyncio库来发送异步HTTP请求.我想timeout在发送每个请求之前等待几毫秒(变量) - 当然 - 异步发送它们,而不是在每个请求发送后等待响应.

我正在做类似以下的事情:

@asyncio.coroutine
def handle_line(self, line, destination):
    print("Inside! line {} destination {}".format(line, destination))
    response = yield from aiohttp.request('POST', destination, data=line,
                               headers=tester.headers)
    print(response.status)
    return (yield from response.read())

@asyncio.coroutine
def send_data(self, filename, timeout):
    destination='foo'
    logging.log(logging.DEBUG, 'sending_data')
    with open(filename) as log_file:
        for line in log_file:
            try:
                json_event = json.loads(line)
            except ValueError as e:
                print("Error parsing json event")
            time.sleep(timeout)
            yield from asyncio.async(self.handle_line(json.dumps(json_event), destination))


loop=asyncio.get_event_loop().run_until_complete(send_data('foo.txt', 1))

我得到的输出(通过打印200个响应)看起来这个代码是同步运行的.我究竟做错了什么？

所以我在我的类中创建了一个名为executor的类变量

executor = ThreadPoolExecutor(100)

而不是拥有函数和方法并使用装饰器,我只是使用以下行来处理我的异步方法中的阻塞任务(如io和哈希创建和......)

result = await to_tornado_future(self.executor.submit(blocking_method, param1, param2)

我决定使用这种风格的原因

1-装饰者性质较慢

2-不需要额外的方法和功能

3-它按预期工作,并在需要之前不创建线程

我对吗 ？请使用原因(我想知道我使用的方式,更慢或使用更多资源或......)

更新

基于Ben回答,我的上述方法不正确所以我最终根据需要使用了以下功能,我认为这是最好的方法

def pool(pool_executor, fn, *args, **kwargs):
    new_future = Future()
    result_future = pool_executor.submit(fn, *args, **kwargs)
    result_future.add_done_callback(lambda f: new_future.set_result(f.result()))
    return new_future

用法:

result = await pool(self.executor, time.sleep, 3)

我写了很多依赖于精确的周期性方法调用的代码.我一直在使用Python的futures库将调用提交到运行时的线程池,并在循环中的调用之间休眠:

executor = ThreadPoolExecutor(max_workers=cpu_count())

def remote_call():
    # make a synchronous bunch of HTTP requests

def loop():
    while True:
        # do work here
        executor.submit(remote_call)
        time.sleep(60*5)

但是,我注意到这个实现在长时间运行后引入了一些漂移(例如,我运行此代码大约10个小时,并注意到大约7秒的漂移).对于我的工作,我需要在确切的秒上运行,毫秒甚至更好.有些人指出我asyncio("火与忘记"python async/await),但我无法在Python 2.7中使用它.

我不是在寻找黑客.我真正想要的是类似于Go time.Tick或Netty的东西HashedWheelTimer.

假设有一个可以进行各种数据库查询的库：

import time

def queryFoo():
    time.sleep(4)
    return "foo"

def queryBar():
    time.sleep(4)
    return "bar"

我想同时执行这两个查询，而不必添加async到方法签名或添加装饰器。这些功能完全不应该依赖于异步。

在其中利用这些非异步功能的最佳方法是什么asyncio？

我正在寻找某种形式的东西：

#I need an 'asyncWrapper'

results = asyncio.gather(asyncWrapper(queryFoo()), asyncWrapper(queryBar()))

预先感谢您的考虑和回应。

是否可以在python中组合异步上下文管理器？类似于asyncio.gather,但能够与上下文管理器一起使用的东西.像这样的东西:

async def foo():
    async with asyncio.gather_cm(start_vm(), start_vm()) as vm1, vm2:
        await vm1.do_something()
        await vm2.do_something()

这目前可能吗？

一旦所有HTTP请求返回，我想启动大量HTTP请求并收集其结果。使用可以以非阻塞方式发送请求asyncio，但是我在收集其结果时遇到问题。

我知道针对此特定问题而设计的解决方案，例如aiohttp。但是HTTP请求只是一个例子，我的问题是如何asyncio正确使用。

在服务器端，我有flask可以localhost/用“ Hello World！” 回答每个请求，但是要等待0.1秒再回答。在所有示例中，我正在发送10个请求。同步代码大约需要1秒钟，异步版本则可以在0.1秒内完成。

在客户端，我想同时启动许多请求并收集其结果。我正在尝试以三种不同的方式来做到这一点。由于asyncio需要执行者来解决阻塞代码，因此所有方法都调用loop.run_in_executor。

这些代码在它们之间共享：

import requests
from time import perf_counter
import asyncio

loop = asyncio.get_event_loop()

async def request_async():
    r = requests.get("http://127.0.0.1:5000/")
    return r.text

def request_sync():
    r = requests.get("http://127.0.0.1:5000/")
    return r.text

方法1：

asyncio.gather()在任务列表中使用，然后单击run_until_complete。阅读Asyncio.gather与asyncio.wait之后，似乎collect会等待结果。但事实并非如此。因此，此代码立即返回，而无需等待请求完成。如果我在此处使用阻止功能，则可以正常工作。为什么我不能使用异步功能？

# approach 1
start = perf_counter()
tasks = []
for i in range(10):
    tasks.append(loop.run_in_executor(None, request_async)) # <---- using async function !

gathered_tasks = asyncio.gather(*tasks)
results = loop.run_until_complete(gathered_tasks) …

我想侦听来自同一对象的多个实例的事件，然后将此事件流合并为一个流。例如，如果我使用异步生成器：

class PeriodicYielder: 
    def __init__(self, period: int) -> None: 
        self.period = period 

    async def updates(self): 
        while True: 
            await asyncio.sleep(self.period)
            yield self.period

我可以成功监听一个实例的事件：

async def get_updates_from_one(): 
    each_1 = PeriodicYielder(1) 
    async for n in each_1.updates(): 
        print(n)
# 1
# 1
# 1
# ...

但是，如何从多个异步生成器获取事件？换句话说：如何按多个异步生成器准备好产生下一个值的顺序进行迭代？

async def get_updates_from_multiple(): 
    each_1 = PeriodicYielder(1) 
    each_2 = PeriodicYielder(2) 
    async for n in magic_async_join_function(each_1.updates(), each_2.updates()): 
        print(n)
# 1
# 1
# 2
# 1
# 1
# 2
# ...

stdlib或3rd party模块中是否有这种magic_async_join_function？

当我转到asyncio页面时，第一个示例是hello world程序。当我在python上运行它时3.73，我看不出与正常的任何区别，有人可以告诉我区别并举一个简单的例子吗？

In [1]: import asyncio
   ...:
   ...: async def main():
   ...:     print('Hello ...')
   ...:     await asyncio.sleep(5)
   ...:     print('... World!')
   ...:
   ...: # Python 3.7+
   ...: asyncio.run(main())
Hello ...
... World!

In [2]:

In [2]: import time
   ...:
   ...: def main():
   ...:     print('Hello ...')
   ...:     time.sleep(5)
   ...:     print('... World!')
   ...:
   ...: # Python 3.7+
   ...: main()
Hello ...
... World!

我故意将时间从1秒增加到5秒，希望看到一些特别的东西，但我没有。