标签: python-asyncio

首先，道歉。我对 Python 还很陌生。我有 Java/C# 编码背景。我在很多方面都喜欢 Python 的简单性，但也发现一些标准很难确定。

例如，我已经成功地运行了一个 Discord 机器人。异步方法运行良好。但我想安排一个作业每（比如）30 分钟运行一次。然而，当我输入 asyncio.run(job()) 时，Python 告诉我“run”不是 asyncio 的属性。我真的不知道为什么会这么说。哎呀，asyncio 是做到这一点的“正确”方法吗？

有没有可能不和谐的导入以某种方式掩盖了它？我想我可能需要买本书什么的！

再次感谢。我确实尝试对此进行搜索，但没有任何结果！

我有一个 http 服务器，它在线程中接收一些请求。该http服务器解析接收到的数据并将部分数据发送到不同的进程。

此进程的响应可能需要一些时间，并通过不同线程中的回调接收。

这里的问题是我需要回调函数中收到的部分响应来响应我收到的初始请求，所以我的想法是为此使用 asyncio.Task

我的代码如下：

my_futures = dict()

class HTTPHandler(http.server.BaseHTTPRequestHandle):
    do_POST(self):
        #parse data
        asyncio.create_task(self.send_response())

    async def send_response(self):
        global my_futures
        my_futures[uuid] = asyncio.get_event_loop().create_future()
        send_data_to_processB()
        await my_future
        self.send_response()

# Callback where I receive the result from process B
def on_message(message):
    global my_futures
    my_futures[message['uuid']].set_result(message['result'])

if __name__ == '__main__':
    server = http.server.HTTPServer(LISTEN_ADDR, MyHTTPHandler)
    threading.Thread(target=server.serve_forever).start()
    processB.register_callback(on_message)

这种方法的问题是任务没有被执行并且回调根本没有收到结果。

我还尝试更改do_POST要使用的方法asyncio.run(self.send_response())而不是asyncio.create_task，使用这种方法我的回调获取结果并设置结果，但协程只是挂在await my_future

我怎样才能完成这个任务？

如果我运行以下代码：

import asyncio
import time
import concurrent.futures

def cpu_bound(mul):
    for i in range(mul*10**8):
        i+=1
    print('result = ', i)
    return i

async def say_after(delay, what):
    print('sleeping async...')
    await asyncio.sleep(delay)
    print(what)

# The run_in_pool function must not block the event loop
async def run_in_pool():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        result = executor.map(cpu_bound, [1, 1, 1])

async def main():
    task1 = asyncio.create_task(say_after(0.1, 'hello'))
    task2 = asyncio.create_task(run_in_pool())
    task3 = asyncio.create_task(say_after(0.1, 'world'))

    print(f"started at {time.strftime('%X')}")
    await task1
    await task2
    await task3
    print(f"finished at {time.strftime('%X')}")

if __name__ …

我必须同时保持我的应用程序与 Python 2 和 3 兼容。

我有一些如下代码，我希望能够异步调用函数，然后等待所有的 future 得到解决。

代码：

import requests

# This would return a future
def get_xhr(url):
    return requests.get('https://www.{}.com'.format(url))

# This would return an array of futures
def get_search_engines():
    urls = ['google', 'yahoo', 'bing']
    return [get_xhr(url) for url in urls]


# Here I want to wait for all the futures to be resolved
get_search_engines()

print('All requests are done')

Asyncio async/await 似乎只与 Python 3 兼容。

能够运行与 python 2/3 兼容的异步函数的最佳方法是什么？

在这个简单的生产者/消费者示例中，就好像await queue.put(item)不释放事件循环以允许消费者运行直到完成。这会导致生产者将其所有项目放入队列中，然后消费者才能将其取出。

这是预期的吗？

await queue.put(item)如果我遵循with ，我就会得到我正在寻找的结果await asyncio.sleep(0)。

然后，生产者将 1 个项目放入队列中，然后消费者从队列中取出 1 个项目。

我在 Python 3.6.8 和 3.7.2 中得到相同的结果。

import asyncio

async def produce(queue, n):
    for x in range(1, n + 1):
        print('producing {}/{}'.format(x, n))
        item = str(x)
        await queue.put(item)
        # await asyncio.sleep(0)
    await queue.put(None)

async def consume(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print('consuming item {}...'.format(item))

loop = asyncio.get_event_loop()
queue = asyncio.Queue(loop=loop)
producer_coro = produce(queue, 10)
consumer_coro = consume(queue)
loop.run_until_complete(asyncio.gather(producer_coro, …

当我阅读aiohttp服务器文档时，我遇到了这个

警告 仅将 add_static() 用于开发。在生产中，静态内容应该由 nginx 或 apache 等 Web 服务器处理。

为什么我们不能用来aiohttp提供静态文件？

我有一个脚本可以检查数十万个提供的网站的状态代码，并且我试图将信号量集成到流程中以加快处理速度。问题是，每当我集成信号量时，我只会得到一个填充有 None 对象的列表，而且我不完全确定为什么。

我主要是从其他来源复制代码，因为我还没有完全理解异步编程，但似乎当我调试时我应该从函数中获取结果，但当我收集结果时出现了问题。我尝试过在循环、收集、确保未来等方面进行杂耍，但似乎没有什么能返回有效的事情列表。

async def fetch(session, url):
    try:
        async with session.head(url, allow_redirects=True) as resp:
            return url, resp.real_url, resp.status, resp.reason
    except Exception as e:
        return url, None, e, 'Error'


async def bound_fetch(sem, session, url):
    async with sem:
        await fetch(session, url)


async def run(urls):
    timeout = 15
    tasks = []

    sem = asyncio.Semaphore(100)
    conn = aiohttp.TCPConnector(limit=64, ssl=False)

    async with aiohttp.ClientSession(connector=conn) as session:
        for url in urls:
            task = asyncio.wait_for(bound_fetch(sem, session, url), timeout)
            tasks.append(task)

        responses = await asyncio.gather(*tasks)

    # responses = [await …

在下面的代码中，我想调用task1和task2，但不期望从这些方法返回，这可能吗？

import asyncio
async def say(something, delay):
  await asyncio.sleep(delay)
  print(something)

loop = asyncio.get_event_loop()
task1 = loop.create_task(say('hi', 1))
task2 = loop.create_task(say('hoi', 2))
loop.run_until_complete(asyncio.gather(task1, task2))

我想处理在 while 循环中进入主队列的某些内容，而无需等待，因为我不需要返回函数，例如伪代码：

import asyncio
async def say(something, delay):
  await asyncio.sleep(delay)
  print(something)

def main():
    while True:
        # search for database news
        # call say asynchronous, but I do not need any return, I just want you to do anything, independent
        time.sleep(1)

我有一个持续监控 API 的函数。基本上，该函数获取数据，解析它，然后将其附加到文件中。然后它会等待 15 分钟并一遍又一遍地执行相同的操作。

我想要的是在后台运行这个循环，这样我就不会阻止其余代码的执行。

我正在尝试让 NATS 订阅者持续收听已发布的消息。

我的发布者是一个可以在浏览器中访问的 API 端点。我的订阅者是一个Python应用程序，它应该永远运行，监听已发布的消息。

我的问题是订户从不打印任何内容。如果我将 run_forever() 更改为loop.close()，它可以工作，但会立即关闭。我知道发布者正在工作，因为我可以看到 NATS 服务器的打印输出。

我正在 docker-compose 中运行所有内容。

我的订阅者：

import asyncio
from nats.aio.client import Client as NATS

async def run(loop):

await nc.connect("nats://nats:4222", loop=loop)

async def message_handler_A(msg):
    print('fsfdsfdsfdsfdsf')

async def message_handler_B(msg):
    print('fdsfdsfdsfdsf')

async def message_handler_C(msg):
    print('sdfdsfdsf')

await nc.subscribe("message_handler_A", cb=message_handler_A)
await nc.subscribe("message_handler_B", cb=message_handler_B)
await nc.subscribe("message_handler_C", cb=message_handler_C)
print('receiving')


if __name__ == '__main__':
    print("RUNNING")

    nc = NATS()
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop) 


    loop.run_until_complete(run(loop))
    loop.run_forever()

我的出版商：

import connexion
import six
import json
import asyncio
from nats.aio.client import Client as NATS …