标签: python-asyncio

我承认自己非常懒惰:我需要相当快地做到这一点,并且无法理解Python3 asyncio模块.(有趣的是,我发现这个提升非常直观.)

我需要读取一个将不时阻塞的文件对象(管道).在此期间,我希望能够以设定的间隔(例如每30分钟)启动另一个活动,无论从文件中读取任何内容的可用性.

任何人都可以使用python3 asyncio帮助我使用骨架吗？(我无法安装第三方模块,如twisted.)

我正在尝试编写基于Asyncio的简单程序和使用ZeroMQ实现的发布/订阅设计模式.出版商有2个协同程序; 一个用于侦听传入的订阅,另一个用于将值(通过HTTP请求获取)发布到订阅者.订户订阅特定参数(在这种情况下为城市名称),并等待该值(该城市的温度).

这是我的代码:

publisher.py

#!/usr/bin/env python

import json
import aiohttp
import aiozmq
import asyncio
import zmq


class Publisher:
    BIND_ADDRESS = 'tcp://*:10000'

    def __init__(self):
        self.stream = None
        self.parameter = ""

    @asyncio.coroutine
    def main(self):
        self.stream = yield from aiozmq.create_zmq_stream(zmq.XPUB, bind=Publisher.BIND_ADDRESS)
        tasks = [
            asyncio.async(self.subscriptions()),
            asyncio.async(self.publish())]
        print("before wait")
        yield from asyncio.wait(tasks)
        print("after wait")

    @asyncio.coroutine
    def subscriptions(self):
        print("Entered subscriptions coroutine")
        while True:
            print("New iteration of subscriptions loop")
            received = yield from self.stream.read()
            first_byte = received[0][0]
            self.parameter = received[0][-len(received[0])+1:].decode("utf-8")
            # Subscribe request
            if first_byte == …

我应该替换while True我的代码(没有asyncio)或者我应该使用asyncio事件循环来完成相同的结果.

目前我在某种与"zeromq"连接的"worker"上工作,接收一些数据,然后对外部工具(服务器)执行一些请求(http).一切都写在正常的阻塞IO中.使用asyncio事件循环摆脱是否有意义while True: ...？

将来它可能会在asyncio中完全重写,但现在我害怕从asyncio开始.

我是asyncio的新手,并不是这个库的所有部分对我来说都很清楚:)

谢谢 :)

使用python 3.5或更高版本,直接应用于await未来或任务与包装它之间有什么区别asyncio.wait_for吗？文档不清楚何时适合使用wait_for,我想知道它是否是旧的基于生成器的库的遗迹.下面的测试程序似乎没有显示出任何差异,但这并没有真正证明什么.

import asyncio

async def task_one():
    await asyncio.sleep(0.1)
    return 1

async def task_two():
    await asyncio.sleep(0.1)
    return 2

async def test(loop):
    t1 = loop.create_task(task_one())
    t2 = loop.create_task(task_two())

    print(repr(await t1))
    print(repr(await asyncio.wait_for(t2, None)))

def main():
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(test(loop))
    finally:
        loop.close()

main()

我的应用程序从慢速i/o源读取数据,进行一些处理,然后将其写入本地文件.我用这样的生成器实现了这个:

import time

def io_task(x):
    print("requesting data for input %s" % x)
    time.sleep(1)   # this simulates a blocking I/O task
    return 2*x

def producer(xs):
    for x in xs:
        yield io_task(x)

def consumer(xs):
    with open('output.txt', 'w') as fp:
        for x in xs:
            print("writing %s" % x)
            fp.write(str(x) + '\n')

data = [1,2,3,4,5]
consumer(producer(data))

现在我想在asyncio的帮助下并行完成这项任务,但我似乎无法弄清楚如何.对我来说,主要的问题是通过生成器直接从生产者向消费者提供数据,同时让asyncio发出多个并行请求io_task(x).而且,整体async def与@asyncio.coroutine事情让我感到困惑.

有人可以告诉我如何构建一个使用asyncio此示例代码的最小工作示例吗？

(注意:只是调用,缓冲结果然后将它们写入文件是不行的io_task().我需要一个可以超出主存的大数据集的解决方案,这就是为什么我一直在使用生成器然而,假设消费者总是比所有生产者总和更快,这是安全的.

大家好，我在尝试理解asyncio和aiohttp并使两者正常工作方面遇到困难。不仅我不正确地了解自己在做什么，这时我遇到了一个我不知道如何解决的问题。

我正在使用Windows 10 64位最新更新。

以下代码使用asyncio返回了标题中Content-Type中不包含html的页面列表。

import asyncio
import aiohttp

MAXitems = 30

async def getHeaders(url, session, sema):
    async with session:
        async with sema:
            try:
                async with session.head(url) as response:
                    try:
                        if "html" in response.headers["Content-Type"]:
                            return url, True
                        else:
                            return url, False
                    except:
                        return url, False
            except:
                return url, False


def checkUrlsWithoutHtml(listOfUrls):
    headersWithoutHtml = set()
    while(len(listOfUrls) != 0):
        blockurls = []
        print(len(listOfUrls))
        items = 0
        for num in range(0, len(listOfUrls)):
            if num < MAXitems:
                blockurls.append(listOfUrls[num - items])
                listOfUrls.remove(listOfUrls[num - items])
                items …

我有一个类,其方法如下所示:

# self.stop_event -> threading.Event
def run(self):
    while not self.stop_event.wait(3):  # i.e. repeat every 3 sec
        pass  # do stuff

这个想法是,其中一些是在他们自己的线程中运行的,并且在某个时刻,一个线程会运行stop_event.set(),这自然会阻止所有其他线程.我想为此切换到asyncio,因为其中的任务run主要是休眠和执行IO.因此,我得:

# self.stop_event -> asyncio.Event
async def run(self):
    while not self.stop_event.is_set():
        await asyncio.sleep(3)
        pass  # do stuff

问题是asyncio.Event无法等待,因此在设置时,在方法完成之前最多等待3秒钟.这是一个问题,因为睡眠时间可能是几分钟.目前,我正在解决这个问题,run包括一个asyncio.Task然后取消它event_loop.call_soon(the_task.cancel).

我想问一下是否有更好的方法来实现上述目标？有没有办法我可以在asyncio.Event某种程度上等待超时,类似于threading.Event？

在下面的代码中：

import asyncio

async def task_func():
    print('in task_func')
    return 'the result'


async def main(loop):
    print('creating task')
    task = loop.create_task(task_func())
    print('waiting for {!r}'.format(task))
    await asyncio.sleep(2)
    return_value = await task
    print('task completed {!r}'.format(task))
    print('return value: {!r}'.format(return_value))


event_loop = asyncio.new_event_loop()
try:
    event_loop.run_until_complete(main(event_loop))
finally:
    event_loop.close()

当我执行代码时，结果如下：

creating task
waiting for `<Task pending coro=<task_func() running at <ipython-input-29-797f29858344>:1>>`
in task_func
task completed `<Task finished coro=<task_func() done, defined at <ipython-input-29-797f29858344>:1> result='the result'>`
return value: 'the result'

但是我不明白您设置的代码何时loop.create_task(task_func())执行。具体来说，我假设当您将一个任务添加到事件循环中时，它很快就会执行，因此我认为in task_func是在之前打印的waiting for <Task...。

然后我发现它 …

是否可以采用阻塞功能,work并使其在ProcessPoolExecutor具有多个工作线程的情况下同时运行？

import asyncio
from time import sleep, time
from concurrent.futures import ProcessPoolExecutor

num_jobs = 4
queue = asyncio.Queue()
executor = ProcessPoolExecutor(max_workers=num_jobs)
loop = asyncio.get_event_loop()

def work():
    sleep(1)

async def producer():
    for i in range(num_jobs):
        results = await loop.run_in_executor(executor, work)
        await queue.put(results)

async def consumer():
    completed = 0
    while completed < num_jobs:
        job = await queue.get()
        completed += 1

s = time()
loop.run_until_complete(asyncio.gather(producer(), consumer()))
print("duration", time() - s)

在具有4个以上内核的计算机上运行上述操作需要大约4秒钟.你怎么写producer这样的,上面的例子只需要约1秒钟？

考虑以下程序。

import asyncio
import multiprocessing
from multiprocessing import Queue
from concurrent.futures.thread import ThreadPoolExecutor
import sys


def main():
    executor = ThreadPoolExecutor()
    loop = asyncio.get_event_loop()
    # comment the following line and the shutdown will work smoothly
    asyncio.ensure_future(print_some(executor))

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        print("shutting down")
        executor.shutdown()
        loop.stop()
        loop.close()
        sys.exit()


async def print_some(executor):
    print("Waiting...Hit CTRL+C to abort")
    queue = Queue()
    loop = asyncio.get_event_loop()
    some = await loop.run_in_executor(executor, queue.get)
    print(some)


if __name__ == '__main__':
    main()

我只想在按“ CTRL + C”时正常关机。但是，执行程序线程似乎阻止了该操作（即使我确实致电shutdown）