标签: python-asyncio

考虑以下程序(在CPython 3.4.0b1上运行):

import math
import asyncio
from asyncio import coroutine

@coroutine
def fast_sqrt(x):
   future = asyncio.Future()
   if x >= 0:
      future.set_result(math.sqrt(x))
   else:
      future.set_exception(Exception("negative number"))
   return future


def slow_sqrt(x):
   yield from asyncio.sleep(1)
   future = asyncio.Future()
   if x >= 0:
      future.set_result(math.sqrt(x))
   else:
      future.set_exception(Exception("negative number"))
   return future


@coroutine
def run_test():
   for x in [2, -2]:
      for f in [fast_sqrt, slow_sqrt]:
         try:
            future = yield from f(x)
            print("\n{} {}".format(future, type(future)))
            res = future.result()
            print("{} result: {}".format(f, res))
         except Exception as e:
            print("{} exception: {}".format(f, …

我asyncio.Protocol.data_received在新的Python asyncio模块的回调中执行异步操作时遇到问题.

考虑以下服务器:

class MathServer(asyncio.Protocol):

   @asyncio.coroutine
   def slow_sqrt(self, x):
      yield from asyncio.sleep(1)
      return math.sqrt(x)

   def fast_sqrt(self, x):
      return math.sqrt(x)

   def connection_made(self, transport):
      self.transport = transport

   #@asyncio.coroutine
   def data_received(self, data):
      print('data received: {}'.format(data.decode()))
      x = json.loads(data.decode())
      #res = self.fast_sqrt(x)
      res = yield from self.slow_sqrt(x)
      self.transport.write(json.dumps(res).encode('utf8'))
      self.transport.close()

与以下客户一起使用:

class MathClient(asyncio.Protocol):

   def connection_made(self, transport):
      transport.write(json.dumps(2.).encode('utf8'))

   def data_received(self, data):
      print('data received: {}'.format(data.decode()))

   def connection_lost(self, exc):
      asyncio.get_event_loop().stop()

随着self.fast_sqrt被召唤,一切都按预期工作.

有self.slow_sqrt,它不起作用.

它也不适用self.fast_sqrt于@asyncio.coroutine装饰器data_received.

我觉得我在这里缺少一些基本的东西.

完整的代码在这里: …

我想使用队列将数据从父进程传递到通过其启动的子进程multiprocessing.Process.但是,由于父进程使用Python的新asyncio库,因此队列方法必须是非阻塞的.据我所知,asyncio.Queue是为了进行任务间通信而不能用于进程间通信.此外,我知道multiprocessing.Queue有put_nowait()和get_nowait()方法,但我实际上需要协程仍然会阻止当前任务(但不是整个过程).有没有办法创建包装put_nowait()/的协同程序get_nowait()？另一方面multiprocessing.Queue,在同一进程中运行的事件循环之后,内部使用的线程是否兼容？

如果没有,我还有其他选择吗？我知道我可以通过使用异步套接字自己实现这样的队列,但我希望我能避免这种情况......

编辑: 我也考虑使用管道而不是套接字,但它似乎asyncio不兼容multiprocessing.Pipe().更确切地说,Pipe()返回一个不是文件Connection对象的对象元组.但是,方法/方法和/都需要类似文件的对象,因此无法异步读取/写入这样的对象.相比之下,包用作管道的通常的文件类对象完全没有问题,并且可以很容易地与之结合使用.asyncio.BaseEventLoopadd_reader()add_writer()connect_read_pipe()connect_write_pipe()Connectionsubprocessasyncio

更新: 我决定进一步探索管道方法:通过检索文件描述符并将其传递给我,将Connection返回的对象转换multiprocessing.Pipe()为类文件对象.最后,我将生成的类文件对象传递给事件循环的/ .(如果有人对确切的代码感兴趣,有一些关于相关问题的邮件列表讨论.)然而,流媒体给了我一个,我没有设法解决这个问题.考虑到缺少对Windows的支持,我不会再继续这样做了.fileno()os.fdopen()connect_read_pipe()connect_write_pipe()read()OSError: [Errno 9] Bad file descriptor

在Windows上使用Python 3.4.1,我发现在执行asyncio事件循环时,我的程序不能被中断(即在终端中按Ctrl + C).更重要的是,SIGINT信号被忽略.相反,我已经确定在不在事件循环中时处理SIGINT.

为什么执行asyncio事件循环时会忽略SIGINT？

下面的程序应该演示问题 - 在终端中运行它并尝试通过按Ctrl + C来停止它,它应该继续运行:

import asyncio
import signal


# Never gets called after entering event loop
def handler(*args):
    print('Signaled')


signal.signal(signal.SIGINT, handler)

print('Event loop starting')
loop = asyncio.SelectorEventLoop()
asyncio.set_event_loop(loop)
loop.run_forever()
print('Event loop ended')

请参阅官方(郁金香)邮件列表上的讨论.

我想使用Python 3 asyncio模块来创建服务器应用程序.我使用主事件循环来监听网络,当收到新数据时,它会进行一些计算并将结果发送给客户端."做一些计算"是否需要新的事件循环？或者它可以使用主事件循环？

我无法绕过Python 3的Asyncio库.我有一个zipcodes列表,我正在尝试对API进行异步调用,以使每个zipcodes对应城市和州.我可以通过for循环顺序成功完成,但我希望在大型邮政编码列表的情况下使其更快.

这是我的原创作品的一个例子

import urllib.request, json

zips = ['90210', '60647']

def get_cities(zipcodes):
    zip_cities = dict()
    for idx, zipcode in enumerate(zipcodes):
        url = 'http://maps.googleapis.com/maps/api/geocode/json?address='+zipcode+'&sensor=true'
        response = urllib.request.urlopen(url)
        string = response.read().decode('utf-8')
        data = json.loads(string)
        city = data['results'][0]['address_components'][1]['long_name']
        state = data['results'][0]['address_components'][3]['long_name']
        zip_cities.update({idx: [zipcode, city, state]})
    return zip_cities

results = get_cities(zips)
print(results)
# returns {0: ['90210', 'Beverly Hills', 'California'],
#          1: ['60647', 'Chicago', 'Illinois']}

这是我尝试使其异步的可怕的非功能性尝试

import asyncio
import urllib.request, json

zips = ['90210', '60647']
zip_cities = dict()

@asyncio.coroutine
def get_cities(zipcodes):
    url = 'http://maps.googleapis.com/maps/api/geocode/json?address='+zipcode+'&sensor=true'
    response …

我不能使用常规工具和技术来测量协程的性能,因为它await不应该考虑它所花费的时间(或者它应该只考虑从等待但不是IO延迟读取的开销).

那么如何衡量协程所需的时间呢？我如何比较2个实现并找到更高效的？我使用什么工具？

这是我的python程序的相关代码:

import discord
import asyncio

class Bot(discord.Client):
    def __init__(self):
        super().__init__()

    @asyncio.coroutine
    def my_background_task(self):
        yield from self.wait_until_ready()
        while not self.is_closed:
            yield from asyncio.sleep(3600*24) # <- This is line 76 where it fails
            doSomething()

bot = Bot()
loop = asyncio.get_event_loop()
try:
    loop.create_task(bot.my_background_task())
    loop.run_until_complete(bot.login('username', 'password'))
    loop.run_until_complete(bot.connect())
except Exception:
    loop.run_until_complete(bot.close())
finally:
    loop.close()

该程序偶尔退出(自己,但不应该),除了以外没有其他错误或警告

Task was destroyed but it is pending!
task: <Task pending coro=<my_background_task() running at bin/discordBot.py:76> wait_for=<Future pending cb=[Task._wakeup()]>>

如何确保程序不会随机退出？我在Xubuntu 15.10上有Python 3.4.3+.

我有两个函数:第一个def_a是异步函数,第二个是def_b常规函数,调用结果def_a作为add_done_callback函数的回调函数.

我的代码看起来像这样:

import asyncio

def def_b(result):
    next_number = result.result()
    # some work on the next_number
    print(next_number + 1)

async def def_a(number):
    await some_async_work(number)
    return number + 1

loop = asyncio.get_event_loop()
task = asyncio.ensure_future(def_a(1))
task.add_done_callback(def_b)
response = loop.run_until_complete(task)
loop.close()

它的工作非常完美.

当第二个函数def_b变为异步时,问题就开始了.现在它看起来像这样:

async def def_b(result):
    next_number = result.result()
    # some asynchronous work on the next_number
    print(next_number + 1)

但是现在我无法将它提供给add_done_callback函数,因为它不是常规函数.

我的问题是 - 它是否可能,如果是异步的,我怎么能提供def_b给add_done_callback函数def_b？

我们可以使用这两个函数来同步运行任何异步函数：

import asyncio
from asgiref.sync import async_to_sync

asyncio.run(asyncio.sleep(1))
async_to_sync(asyncio.sleep)(1)

有什么不同？我们可以总是使用asyncio.run代替async_to_sync吗？