标签: python-asyncio

在Windows上的Python 3.4下,我需要通过子进程流式传输写入stdout/stderr的数据,即使用Python 3.4中引入的asyncio框架接收其输出.之后我还必须确定程序的退出代码.我怎样才能做到这一点？

https://docs.python.org/3/library/asyncio-eventloop.html#delayed-calls

注意:超时(相对延迟或绝对时间)不应超过一天.

有没有人对这个限制存在的原因有一个很好的解释？

提前致谢

我正在使用py.test为python3代码构建一些测试。该代码使用aiopg（到Postgres的基于Asyncio的接口）访问Postgresql数据库。

我的主要期望：

• 每个测试用例都应该可以访问新的asyncio事件循环。

• 运行时间过长的测试将因超时异常而停止。

• 每个测试用例都应该有权访问数据库连接。

• 在编写测试用例时，我不想重复自己。

使用py.test固定装置，我可以很接近我想要的东西，但是在每个异步测试用例中，我仍然不得不重复自己一遍。

这是我的代码的样子：

@pytest.fixture(scope='function')
def tloop(request):
    # This fixture is responsible for getting a new event loop
    # for every test, and close it when the test ends.
    ...

def run_timeout(cor,loop,timeout=ASYNC_TEST_TIMEOUT):
    """
    Run a given coroutine with timeout.
    """
    task_with_timeout = asyncio.wait_for(cor,timeout)
    try:
        loop.run_until_complete(task_with_timeout)
    except futures.TimeoutError:
        # Timeout:
        raise ExceptAsyncTestTimeout()


@pytest.fixture(scope='module')
def clean_test_db(request):
    # Empty the test database.
    ...

@pytest.fixture(scope='function')
def udb(request,clean_test_db,tloop):
    # Obtain a connection to the database using aiopg
    # (That's why we …

我正在开发一个主要由服务组成的应用程序,这些服务是具有自定义运行循环的线程.

其中一个服务需要产生子进程,我真的不明白它是否有效.官方文件含糊不清.即它说都asyncio supports running subprocesses from different threads和An event loop must run in the main thread在同一节.

如果事件循环必须在主线程中运行,那么甚至可以从不同的线程运行子进程？

我已经使用过像Twisted和Tornado这样的python异步框架.另外我知道python已经通过asyncio模块实现了异步调用的本机实现.我认为(线程,多处理)和异步调用是不同的概念.但不久前我观看了几个与线程和多处理相关的视频,似乎所有这些异步人员都在他们之上构建.这是真的吗？

我目前正在使用Python 3.5中的asyncio执行我的第一步,并且有一个问题困扰着我.显然我还没有完全理解协程......

这是我正在做的简化版本.

在我的类中,我有一个open()方法,可以创建一个新线程.在该线程中,我创建了一个新的事件循环和一些主机的套接字连接.然后我让循环永远运行.

def open(self):
    # create thread
    self.thread = threading.Thread(target=self._thread)
    self.thread.start()
    # wait for connection
    while self.protocol is None:
        time.sleep(0.1)

def _thread(self):
    # create loop, connection and run forever
    self.loop = asyncio.new_event_loop()
    coro = self.loop.create_connection(lambda: MyProtocol(self.loop),
                                       'somehost.com', 1234)
    self.loop.run_until_complete(coro)
    self.loop.run_forever()

停止连接现在非常简单,我只是从主线程停止循环:

loop.call_soon_threadsafe(loop.stop)

不幸的是我需要做一些清理工作,特别是在断开与服务器的连接之前我需要清空一个队列.所以我尝试了MyProtocol中的这个stop()方法:

class MyProtocol(asyncio.Protocol):
    def __init__(self, loop):
        self._loop = loop
        self._queue = []

    async def stop(self):
        # wait for all queues to empty
        while self._queue:
            await asyncio.sleep(0.1)
        # disconnect
        self.close()
        self._loop.stop()

队列从协议的data_received()方法中清空,所以我只想等待使用带有asyncio.sleep()调用的while循环.然后我关闭连接并停止循环.

但是如何从主线程调用此方法并等待它？我尝试了以下,但它们似乎都没有工作(协议是当前使用的MyProtocol实例):

loop.call_soon_threadsafe(protocol.stop)
loop.call_soon_threadsafe(functools.partial(asyncio.ensure_future, protocol.stop(), loop=loop)) …

哪里

这是在Linux,Python 3.5.1上.

什么

我正在开发一个监控程序asyncio,其任务在不同的地方await在asyncio.sleep不同时间的电话.

有一些时间点,我希望能够中断所有所述asyncio.sleep呼叫并让所有任务正常进行,但我找不到如何做到这一点.一个示例是正常关闭监视器进程.

如何(失败的假设)

我以为我可以发送一个ALRM信号,但过程就会消失.我尝试用以下方法捕获ALRM信号:

def sigalrm_sent(signum, frame):
    tse.logger.info("got SIGALRM")

signal.signal(signal.SIGALRM, sigalrm_sent)

然后我得到关于捕获SIGALRM的日志行,但是asyncio.sleep调用没有中断.

如何(kludge)

此时,我asyncio.sleep通过调用此协程替换了所有调用:

async def interruptible_sleep(seconds):
    while seconds > 0 and not tse.stop_requested:
        duration = min(seconds, tse.TIME_QUANTUM)
        await asyncio.sleep(duration)
        seconds -= duration

因此,我只需要选择一个TIME_QUANTUM不太小也不太大的东西.

但

有没有办法打断所有正在运行的asyncio.sleep呼叫,我错过了它？

我想知道如何使用asyncio来处理类似于nodeJS的任务.我想在不打开线程的情况下同时运行任务.例:

import asyncio

@asyncio.coroutine
def my_coroutine(task_name, seconds_to_sleep=3):
    print('{0} sleeping for: {1} seconds'.format(task_name, seconds_to_sleep))
    yield from asyncio.sleep(seconds_to_sleep)
    print('{0} is finished'.format(task_name))


loop = asyncio.get_event_loop()
tasks = [
    my_coroutine('task1', 4),
    my_coroutine('task2', 3),
    my_coroutine('task3', 2)]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

将输出:

task1 sleeping for: 4 seconds
task2 sleeping for: 3 seconds
task3 sleeping for: 2 seconds
task3 is finished
task2 is finished
task1 is finished

但是当我尝试用不同的任务完成它时,它将无法正常工作.

import asyncio
import timeit

@asyncio.coroutine
def my_coroutine(task_name):
    print('order placed for ' + task_name)
    print(timeit.timeit('1 + 3 ', number=50000000))
    print(task_name + ' …

我有以下功能正常,但由于某种原因,请求似乎是同步执行,而不是异步.

我现在的假设是,由于for record in recordsmain函数中的for循环,这种情况正在发生,但我不知道如何更改它以便请求可以执行异步.如果不是这样,我还需要改变什么呢？

async def do_request(query_string):
        base_url = 'https://maps.googleapis.com/maps/api/place/textsearch/json?'
        params = {'key': google_api_key,
                  'query': query_string}
        async with aiohttp.ClientSession() as session:
            async with session.request('GET', base_url, params=params) as resp:
                return resp


async def main():
    create_database_and_tables()
    records = prep_sample_data()[:100]

    for record in records:
        r = Record(record)

        if not r.is_valid:
            continue

        query_string = r.generate_query_string()

        resp = await do_request(query_string)
        print("NOW WRITE TO DATABASE")

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

我目前正在使用python3 asyncio中的子流程执行任务。我的代码只是简单地写入stdin并同时读取stdout / stderr：

import asyncio


async def read_stdout(stdout):
    print('read_stdout')
    while True:
        buf = await stdout.read(10)
        if not buf:
            break

        print(f'stdout: { buf }')


async def read_stderr(stderr):
    print('read_stderr')
    while True:
        buf = await stderr.read()
        if not buf:
            break

        print(f'stderr: { buf }')


async def write_stdin(stdin):
    print('write_stdin')
    for i in range(100):
        buf = f'line: { i }\n'.encode()
        print(f'stdin: { buf }')

        stdin.write(buf)
        await stdin.drain()
        await asyncio.sleep(0.5)


async def run():
    proc = await asyncio.create_subprocess_exec(
        '/usr/bin/tee',
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE)

    await asyncio.gather(
        read_stderr(proc.stderr),
        read_stdout(proc.stdout), …