标签: python-asyncio

我正在使用asyncio作为网络框架.

在下面的代码中(low_level是我们的低级函数,mainblock是我们的程序入口,user_func是用户定义的函数):

import asyncio

loop = asyncio.get_event_loop()
""":type :asyncio.AbstractEventLoop"""


def low_level():
    yield from asyncio.sleep(2)


def user_func():
    yield from low_level()


if __name__ == '__main__':
    co = user_func()
    loop.run_until_complete(co)

我想包装low_level正常函数而不是coroutine(compatibility等等),但是low_level在事件循环中.如何将其包装为正常功能？

我想在Executor中使用asyncio调用loop.run_in_executor启动阻塞函数,然后稍后取消它,但这对我来说似乎不起作用.

这是代码:

import asyncio
import time

from concurrent.futures import ThreadPoolExecutor


def blocking_func(seconds_to_block):
    for i in range(seconds_to_block):
        print('blocking {}/{}'.format(i, seconds_to_block))
        time.sleep(1)

    print('done blocking {}'.format(seconds_to_block))


@asyncio.coroutine
def non_blocking_func(seconds):
    for i in range(seconds):
        print('yielding {}/{}'.format(i, seconds))
        yield from asyncio.sleep(1)

    print('done non blocking {}'.format(seconds))


@asyncio.coroutine
def main():
    non_blocking_futures = [non_blocking_func(x) for x in range(1, 4)]
    blocking_future = loop.run_in_executor(None, blocking_func, 5)
    print('wait a few seconds!')
    yield from asyncio.sleep(1.5)

    blocking_future.cancel()
    yield from asyncio.wait(non_blocking_futures)



loop = asyncio.get_event_loop()
executor = ThreadPoolExecutor(max_workers=1)
loop.set_default_executor(executor)
asyncio.async(main())
loop.run_forever()

我希望上面的代码只允许阻塞函数输出:

blocking 0/5
blocking …

我对如何asyncio在Python 3.4中使用该模块感到困惑.我有一个searching搜索引擎的API,并希望每个搜索请求可以并行或异步运行,这样我就不必等待一次搜索完成另一次搜索.

这是我的高级搜索API,用于使用原始搜索结果构建一些对象.搜索引擎本身正在使用某种asyncio机制,所以我不会打扰它.

# No asyncio module used here now
class search(object):
  ...
  self.s = some_search_engine()
  ...
  def searching(self, *args, **kwargs):
    ret = {}
    # do some raw searching according to args and kwargs and build the wrapped results
    ...
    return ret

为了尝试异步请求,我编写了以下测试用例来测试如何将我的东西与asyncio模块进行交互.

# Here is my testing script
@asyncio.coroutine
def handle(f, *args, **kwargs):
  r = yield from f(*args, **kwargs)
  return r

s = search()
loop = asyncio.get_event_loop()
loop.run_until_complete(handle(s.searching, arg1, arg2, ...))
loop.close()

通过运行pytest, …

我们假设我有以下代码:

import asyncio
import threading

queue = asyncio.Queue()

def threaded():
    import time
    while True:
        time.sleep(2)
        queue.put_nowait(time.time())
        print(queue.qsize())

@asyncio.coroutine
def async():
    while True:
        time = yield from queue.get()
        print(time)

loop = asyncio.get_event_loop()
asyncio.Task(async())
threading.Thread(target=threaded).start()
loop.run_forever()

这段代码的问题在于asynccoroutine 中的循环永远不会完成第一次迭代,而queue大小正在增加.

为什么会这样发生,我该怎么做才能解决它？

我无法摆脱单独的线程,因为在我的实际代码中,我使用单独的线程与串行设备进行通信,而我还没有找到使用它的方法asyncio.

我想用的ThreadPoolExecutor从蟒蛇协程,委托一些阻塞网络调用到一个单独的线程.但是,运行以下代码:

from concurrent.futures import ThreadPoolExecutor
import asyncio

def work():
  # do some blocking io
  pass

async def main():
  executor = ThreadPoolExecutor()
  await executor.submit(work)

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

导致错误:

TypeError: object Future can't be used in 'await' expression

是不是Future对象等待？为什么说它们不是？

我怎样才能返回await一个Future对象executor.submit？

Python 3.5.0

编辑

使用executor.submit不是我的决定.这在几个库内部使用,比如requests-futures.我正在寻找一种方法来与协同程序中的那些模块互操作.

我正在尝试使用asyncio来进行异步客户端/服务器设置.

出于某种原因,我AssertionError: yield from wasn't used with future在运行客户端时得到了.

搜索此错误并未发生太多变化.
这个错误意味着什么,是什么导致它？

#!/usr/bin/env python3

import asyncio
import pickle
import uuid

port = 9999

class ClientProtocol(asyncio.Protocol):
    def __init__(self, loop):
        self.loop = loop
        self.conn = None
        self.uuid = uuid.uuid4()
        self.other_clients = []

    def connection_made(self, transport):
        print("Connected to server")
        self.conn = transport

        m = "hello"
        self.conn.write(m)

    def data_received(self, data):
        print('Data received: {!r}'.format(data))


    def connection_lost(self, exc):
        print('The server closed the connection')
        print('Stop the event loop')
        self.loop.stop()



# note that in my use-case, main() is …

我正在尝试集成一个tqdm进度条来监视aiohttp在Python 3.5中生成的POST请求.我有一个工作进度条,但似乎无法使用收集结果as_completed().指针感激不尽.

我发现的示例建议使用以下模式,该模式与Python 3.5 async def定义不兼容:

for f in tqdm.tqdm(asyncio.as_completed(tasks), total=len(coros)):
    yield from f

在没有进度条的情况下工作(虽然部分编辑)异步代码:

def async_classify(records):

    async def fetch(session, name, sequence):
        url = 'https://app.example.com/api/v0/search'
        payload = {'sequence': str(sequence)}
        async with session.post(url, data=payload) as response:
            return name, await response.json()

    async def loop():
        auth = aiohttp.BasicAuth(api_key)
        conn = aiohttp.TCPConnector(limit=100)
        with aiohttp.ClientSession(auth=auth, connector=conn) as session:
            tasks = [fetch(session, record.id, record.seq) for record in records]
            responses = await asyncio.gather(*tasks)    
        return OrderedDict(responses)

这是我修改的失败尝试loop():

async def …

两个aiohttp.web.Application()对象可以在同一个进程中运行,例如在不同的端口上运行吗？

我看到一堆aiohttp代码示例:

from aiohttp import web
app = web.Application()
app.router.add_get('/foo', foo_view, name='foo')
web.run_app(app, host='0.0.0.0', port=10000)

我想知道是否有一些等价物web.Applications()可以配置多个同时运行.就像是:

from aiohttp import web
app1 = web.Application()
app1.router.add_get('/foo', foo_view, name='foo')
app2 = web.Application()
app2.router.add_get('/bar', bar_view, name='bar')
# This is the wishful thinking code:
web.configure_app(app1, host='0.0.0.0', port=10000)
web.configure_app(app2, host='0.0.0.0', port=10001)
web.run_apps()

我的用例是我有一个现有的python web框架来做这种事情,我正在构建一个类似于python 3.6和aiohttp的原型.

我知道多个python服务器可以运行在例如nginx之后(另请参阅http://aiohttp.readthedocs.io/en/stable/deployment.html); 那不是我追求的.我想探索两个具有相同asyncio事件循环的aiohttp Web服务器的可能性,在同一个python进程中运行,在两个不同的端口上提供服务.

文档说要重用ClientSession:

不要为每个请求创建会话.很可能每个应用程序都需要一个会话来完成所有请求.

会话内部包含连接池,连接重用和保持活动(两者都默认打开)可以加快总体性能.1

但是在文档中似乎没有任何解释如何做到这一点？有一个例子可能是相关的,但它没有说明如何在其他地方重用池:http://aiohttp.readthedocs.io/en/stable/client.html#keep-alive-connection-pooling-and-cookie-分享

这样的事情是正确的做法吗？

@app.listener('before_server_start')
async def before_server_start(app, loop):
    app.pg_pool = await asyncpg.create_pool(**DB_CONFIG, loop=loop, max_size=100)
    app.http_session_pool = aiohttp.ClientSession()


@app.listener('after_server_stop')
async def after_server_stop(app, loop):
    app.http_session_pool.close()
    app.pg_pool.close()


@app.post("/api/register")
async def register(request):
    # json validation
    async with app.pg_pool.acquire() as pg:
        await pg.execute()  # create unactivated user in db
        async with app.http_session_pool as session:
            # TODO send activation email using SES API
            async with session.post('http://httpbin.org/post', data=b'data') as resp:
                print(resp.status)
                print(await resp.text())
        return HTTPResponse(status=204)

我正在使用第三方模块从API检索数据.我只是想异步等待模块返回有时需要几秒钟的数据并冻结我的应用程序.但是,当我尝试等待对该模块的调用时,我收到TypeError:

TypeError: object dict can't be used in 'await' expression

import thirdPartyAPIwrapper

async def getData():
    retrienveData = await thirdPartyAPIWrapper.data()
    return await retrieveData

def main():
    loop = asncio.get_event_loop()
    data = loop.run_until_complete(getData())
    loop.close
    return data

为什么我不等待类型('dict')？有没有解决的办法？如果与asyncio的异步/等待不能与不返回协程的第三方模块一起工作那么我的其他选择是什么？