标签: python-asyncio

我有一个 python 3.4.3、postgreSQL 9.4、aiopg-0.7.0。多线程应用程序的示例取自该站点。如何使用游泳池？选择操作时线程挂起。

import time
import asyncio
import aiopg
import functools
from threading import Thread, current_thread, Event
from concurrent.futures import Future

class B(Thread):
   def __init__(self, start_event):
       Thread.__init__(self)
       self.loop = None
       self.tid = None
       self.event = start_event

   def run(self):
       self.loop = asyncio.new_event_loop()
       asyncio.set_event_loop(self.loop)
       self.tid = current_thread()
       self.loop.call_soon(self.event.set)
       self.loop.run_forever()

   def stop(self):
       self.loop.call_soon_threadsafe(self.loop.stop)

   def add_task(self, coro):
       """this method should return a task object, that I
         can cancel, not a handle"""
      def _async_add(func, fut):
          try:
              ret = func()
              fut.set_result(ret)
          except Exception as e: …

以下面的程序为例：

import asyncio
from concurrent.futures import ProcessPoolExecutor


def process():
    print('processed')

async def main(loop, executor):
    await loop.run_in_executor(executor, process)
    await asyncio.sleep(60.0)

executor = ProcessPoolExecutor()
loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main(loop, executor))
except KeyboardInterrupt:
    pass
finally:
    executor.shutdown()

如果我在程序运行时点击Ctrl+ C，我会在它退出时得到一个真正的消息回溯：

processed
^CProcess Process-3:
Process Process-4:
Process Process-2:
Traceback (most recent call last):
  File "/usr/local/Cellar/python3/3.5.0/Frameworks/Python.framework/Versions/3.5/lib/pytho
n3.5/multiprocessing/process.py", line 254, in _bootstrap
    self.run()
  File "/usr/local/Cellar/python3/3.5.0/Frameworks/Python.framework/Versions/3.5/lib/pytho
n3.5/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/Cellar/python3/3.5.0/Frameworks/Python.framework/Versions/3.5/lib/pytho
n3.5/concurrent/futures/process.py", line 169, in _process_worker
    call_item = call_queue.get(block=True)
  File "/usr/local/Cellar/python3/3.5.0/Frameworks/Python.framework/Versions/3.5/lib/pytho
n3.5/multiprocessing/queues.py", …

考虑以下代码（解释如下）：

import asyncio

class MyClass(object):
    def __init__(self):
        self.a = 0
    def incr(self, data):
        self.a += 1
        print(self.a)

class GenProtocol(asyncio.SubprocessProtocol):
    def __init__(self, handler, exit_future):
        self.exit_future = exit_future
        self.handler = handler

    def pipe_data_received(self, fd, data):
        if fd == 1:
            self.handler(data)
        else:
            print('An error occurred')

    def process_exited(self):
        self.exit_future.set_result(True)

def start_proc(stdout_handler, *command):
    loop = asyncio.get_event_loop()
    exit_f = asyncio.Future(loop=loop)
    subpr = loop.subprocess_exec(lambda: GenProtocol(stdout_handler, exit_f),
                                 *command,
                                 stdin=None)
    transport, protocol = loop.run_until_complete(subpr)

    @asyncio.coroutine
    def waiter(exit_future):
        yield from exit_future

    return waiter, exit_f

def main():
    my_instance = MyClass()
    loop …

很简单，我只想将来自aiohttp异步 HTTP 请求的响应与标识符（例如字典键）相关联，以便我知道哪个响应对应于哪个请求。

例如，下面的URI调用功能后缀的字典值1，2和3。如何修改它以返回与每个结果关联的键？我只需要能够跟踪哪个请求是哪个......对于熟悉的人来说无疑是微不足道的asyncio

import asyncio
import aiohttp

items = {'a': '1', 'b': '2', 'c': '3'}

def async_requests(items):
    async def fetch(item):
        url = 'http://jsonplaceholder.typicode.com/posts/'
        async with aiohttp.ClientSession() as session:
            async with session.get(url + item) as response:
                return await response.json()

    async def run(loop):
        tasks = []
        for k, v in items.items():
            task = asyncio.ensure_future(fetch(v))
            tasks.append(task)
        responses = await asyncio.gather(*tasks)
        print(responses)

    loop = asyncio.get_event_loop()
    future = asyncio.ensure_future(run(loop))
    loop.run_until_complete(future)

async_requests(items)

输出（缩写）：

[{'id': 2, ...}, …

我可以写一个描述符，返回一个可以等待的未来。

class AsyncDescriptor:
    def __get__(self, obj, cls=None):
         # generate some async future here
         return future

    def __set__(self, obj, value):
         # generate some async future here
         return future

class Device:
    attr=AsyncDescriptor()

device=Device()

现在，我可以用来获得协程中的值value=await device.attr。

我将如何设置此属性？

• await device.attr=5 -> SyntaxError：无法分配给等待表达式
• await setattr(device, 'attr', 5) -> TypeError：对象NoneType不能在'await'表达式中使用
• device.attr=5 -> RuntimeWarning：从未等待协程'__set__'

我有一个库，可以让我选择在外部作业完成时安排回调。使用此回调是否安全Future.set_result()？如果没有，实现这一目标的正确方法是什么？Future 的文档说它的方法不是线程安全的，所以我认为这可能有问题。

我的目标是使用来自 PyOpenCL 的 OpenCL 事件作为 asyncio 代码中的可等待对象。我在考虑这样的辅助函数：

def wrap_opencl_event(event):
    f = asyncio.Future()
    event.set_callback(pyopencl.command_execution_status.COMPLETE, lambda x: f.set_result(None))
    return f

并以这种方式使用它：

async def do_slow_stuff():
    ev1 = pyopencl.enqueue_something()
    ev2 = pyopencl.enqueue_something_else(wait_for=[ev1])
    await wrap_opencl_event(ev2)
    process_results()

好像asyncio.Queue只能由同一个线程推读吗？例如：

import asyncio
from threading import Thread
import time

q = asyncio.Queue()

def produce():
    for i in range(100):
        q.put_nowait(i)
        time.sleep(0.1)

async def consume():
    while True:
        i = await q.get()
        print('consumed', i)

Thread(target=produce).start()
asyncio.get_event_loop().run_until_complete(consume())

只打印

consumed 0

然后挂起。我错过了什么？

迷失在 asyncio 中。

我正在同时学习 Kivy 和 asyncio，并且陷入了解决运行 Kivy 和运行 asyncio 循环的问题上，因为无论我如何转动它，两者都是阻塞调用并且需要按顺序执行（好吧，我希望我错了），例如

loop = asyncio.get_event_loop()
loop.call_soon(MyAsyncApp().run())
loop.run_forever()

我当前的尝试导致应用程序启动，但没有执行协程，例如，当我单击“连接”按钮时，我应该使用 开始调度和执行任务loop.call_soon，但没有任何反应。

有人可以看看我的代码并提出解决问题的正确方法吗？

import asyncio
import random
import time
from kivy.app import App
from kivy.lang import Builder

ui = Builder.load_string('''
BoxLayout:
    orientation: 'vertical'
    GridLayout:
        rows: 2
        cols: 2
        Label:
            text: 'Status:'
            size_hint: 0.3, 1
        Label:
            id: status
            text: ''
        Label:
            text: 'Data:'
            size_hint: 0.7, 1
        Label:
            id: data
            text: ''
    BoxLayout:
        direction: 'horizontal'
        Button:
            text: 'Get Data'
            on_press: app.connect()
        Button:
            text: 'Stop …

从事件线程外部将协程推送到事件线程的pythonic方法是什么？

一直试图弄清楚我返回数据后如何访问我的数据。如果我打印它，这是 x 的示例：

<_GatheringFuture finished result=[[{'ip': '173.245.203.70'}], [{'ip': '196.52.2.82'}], [{'ip': '69.161.4.1'}], [{'ip':'73.180.140.205'}], [{'ip': '96.242.181.128'}], [{'ip': '108.208.74.183'}], ...]>

我尝试了 x[0]、x.result、x.results 等……但一直出现错误。有人知道如何访问 Future 结果中的数据吗？感谢您的时间！下面是完整的代码：

import asyncio
from bs4 import BeautifulSoup
from aiohttp import ClientSession

HEADERS = {'user-agent': 'Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36'}
proxy = 'xxx.xxx.xxx.xxx:xxxxx'

async def fetch(url, session):
 while True:
  x = []
  try:
   async with session.get(url, headers=HEADERS, proxy="http://{}".format(proxy), timeout=9) as r:
    print (r.status)
    if r.status == 200:
     data = await r.read()
     soup = …