标签: python-asyncio

我正在运行需要时间才能完成的功能。用户可以选择停止此功能/事件。有没有简单的方法来停止线程或循环？

class ThreadsGenerator:
    MAX_WORKERS = 5

    def __init__(self):
        self._executor = ThreadPoolExecutor(max_workers=self.MAX_WORKERS)
        self.loop = None
        self.future = None

    def execute_function(self, function_to_execute, *args):
        self.loop = asyncio.get_event_loop()
        self.future = self.loop.run_in_executor(self._executor, function_to_execute, *args)

        return self.future

我想在用户单击停止按钮时尽快停止该功能，而不是等待完成其工作。

提前致谢！

loop = asyncio.get_event_loop()
executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
tasks = [loop.run_in_executor(executor, self.parse_url, url) for url in urls]
await asyncio.wait(tasks, timeout=5)
for (i, task) in zip(urls, tasks):
    result = task.result() if task._state.lower() == 'finished' else []

在上面的代码中，我使用 '_state' 属性来检查任务是否确实完成并且没有超时。

有没有更清洁的方法？我认为我不应该像那样访问 '_state' 属性。

我有一段简单的代码让我发疯了一段时间。几天前我发布了这个问题，询问create_task是否与input. 现在我想出了一些与此相关的事情。我在一个单独的线程中运行事件循环并在其中推送任务。非常直接的代码。

import asyncio
import threading


async def printer(message):
    print(f'[printer] {message}')


def loop_runner(loop):
    loop.run_forever()


if __name__ == '__main__':
    event_loop = asyncio.get_event_loop()
    t = threading.Thread(target=loop_runner, args=(event_loop,))
    t.start()

    for m in ['hello', 'world', 'foo', 'bar']:
        print(f'[loop running ?] {event_loop.is_running()}')
        event_loop.create_task(printer(m))

除了这些日志消息外，什么都不会打印。

[loop running ?] True
[loop running ?] True
[loop running ?] True
[loop running ?] True

现在，如果我在事件循环线程中阻塞并让它在像这样暂停后运行。

def loop_runner(loop):
    time.sleep(1 / 1000)
    loop.run_forever()

一切正常，这被打印出来

[loop running ?] False
[loop running ?] False
[loop running ?] …

我需要从异步调用中收集结果并将它们返回给一个普通函数——桥接异步和同步代码部分让我感到困惑。

首先尝试，使用丑陋的 inout 参数。

import asyncio
import aiohttp

async def one_call(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            txt = await response.text()
            return txt[0:20]


async def do_all(result_inout):
    urls = ["https://cnn.com", "https://nyt.com", "http://reuters.com"]
    out = await asyncio.gather(*[one_call(url) for url in urls])
    result_inout += out

if __name__ == "__main__":
    result_inout = []
    asyncio.run(do_all(result_inout))

    print(result_inout)

第二次尝试，但直接使用事件循环，不鼓励应用程序代码。

if __name__ == "__main__":
    # same imports, same one_call, same urls
    loop = asyncio.get_event_loop()

    aggregate_future = asyncio.gather(*[one_call(url) for url in urls])
    results …

我有一个服务网络套接字的类，并听 PostgreSQL。使用 asyncpg，当我尝试使用 add_listener 时，出现错误：RuntimeWarning: coroutine was never awaited。如何异步/等待回调。我尝试添加“等待 self.listener”，但它不起作用。

有没有办法以另一种方式处理这个问题？

import asyncio
import http
import websockets
import asyncpg

class App(object):

    def __init__(self, loop):
        self.loop = loop
        self.ws_list = []
        self.conn = None

    async def ws_handler(self, ws, path):
        if self.conn is None:
            self.conn = await asyncpg.connect(user='xxx', password='xxx', database='pgws', host='127.0.0.1')
            await self.conn.add_listener('todo_updates', self.listener)
        print('new socket!!!')
        self.ws_list.append(ws)
        while True:
            await asyncio.sleep(1)

    async def listener(self, conn, pid, channel, payload):
        print(payload)
        for ws in self.ws_list:
            task = asyncio.create_task()
            await ws.send(payload)

if __name__ == "__main__": …

我想使用asyncio.gather() “并行”运行一些任务：

import asyncio

async def foo():
    return 42

async def main():
    results = await asyncio.gather(foo() for i in range(10))
    print(results)

asyncio.run(main())

但它失败了RuntimeError: Task got bad yield：

/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/events.py:88: RuntimeWarning: coroutine 'foo' was never awaited
  self._context.run(self._callback, *self._args)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py", line 583, in run_until_complete
    return future.result()
  File "<stdin>", line 2, in …

我希望传递params和headers到aiohttp.ClientSession如图所示这里。

这是我尝试过的：

    async def make_request(self, url, headers, params):
        async with aiohttp.ClientSession(headers=headers, params=params) as session:
            async with self.limit, session.get(url=url) as response:
                await asyncio.sleep(self.rate)
                resp = await response.read()
                return resp

async def process(url, url_id, update_id, rate, limit):
    limit = asyncio.BoundedSemaphore(limit)

    f = Fetch(
        rate=rate,
        limit=limit,
    )

    if "coinmarketcap" in url:
        params = {
            'start': '1',
            'limit': '1',
            'convert': 'USD,BTC'
        }
        headers = {
            'Accepts': 'application/json',
            'X-CMC_PRO_API_KEY': API_KEY,
        }
    else:
        params = {}
        headers = …