标签: python-asyncio

我正在尝试在 的帮助下同时发送 POST 请求concurrent.futures。由于某种原因，我无法设置自定义标头。我想设置

1. Authorization
2. Content-type

这是我到目前为止所取得的进展。

import asyncio
import concurrent.futures
import requests
from urllib.parse import urlencode, quote_plus


params = urlencode({'a': 'b', 'c': 'd', 'e': 'f'})
headers = {"Content-type": "application/x-www-form-urlencoded","Accept": "*/*","Authorization": "Bearer kdjalskdjalskd"}

async def main():

    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:

        loop = asyncio.get_event_loop()
        futures = [
            loop.run_in_executor(
                executor, 
                requests.post,
                'https://fac03c95.ngrok.io',params, headers)
            for i in range(20)
        ]
        for response in await asyncio.gather(*futures):
            print(response.text)
            pass

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

但由于某种原因，标头似乎没有显示在请求中。有人可以帮我从这里出去吗？

提前致谢 ：）

我已经熟悉Python，但我刚刚开始学习“异步编程”和“io任务”主题。在教程中，我看到它asyncio.Queue()与生产者和消费者相关的东西一起使用，但我不太理解它。

有人可以简化一下，并举几个例子吗？

我目前正在构建一个项目，需要向各个端点发出多个请求。我将这些请求包装在 Aiohttp 中以允许异步。

问题：
我有三个队列：queue1、queue2和queue3。此外，我还有三个工作函数（worker1、worker2、worker3），它们与各自的队列关联。第一个队列立即填充运行前已知的列表 ID。当请求完成并将数据提交到数据库时，它将 ID 传递给queue2。Aworker2将获取此 ID 并请求更多数据。根据这些数据，它将开始生成 ID 列表（与 中的 ID 不同）queue1/queue2。worker2会将 ID 放入 中queue3。最后，在提交到数据库之前，worker3将从中获取此 ID并请求更多数据。queue3

问题queue.join()是由于阻塞调用而出现的。每个工作线程都绑定到一个单独的队列，因此连接queue1将阻塞，直到完成。这很好，但它也违背了使用异步的目的。如果不使用join()该程序，则无法检测队列何时完全为空。另一个问题是，当其中一个队列为空但仍有数据尚未添加时，可能会出现静默错误。

基本代码概要如下：

queue1 = asyncio.Queue()
queue2 = asyncio.Queue()
queue3 = asyncio.Queue()

async with aiohttp.ClientSession() as session:
    for i in range(3):
        tasks.append(asyncio.create_task(worker1(queue1)))

    for i in range(3):
        tasks.append(asyncio.create_task(worker2(queue2)))

    for i in range(10): …

我正在尝试创建一个不和谐机器人，我需要在另一个新线程中运行异步函数，因为主线程需要运行另一个函数（不和谐客户端）

我正在努力实现的目标：

# This methods needs to run in another thread
async def discord_async_method():
    while True:
        sleep(10)
        print("Hello World")
        ... # Discord Async Logic

# This needs to run in the main thread
client.run(TOKEN)

thread = ""

try:
    # This does not work, throws error "printHelloWorld Needs to be awaited"
    thread = Thread(target=discord_async_method)
    thread.start()
except (KeyboardInterrupt, SystemExit):

    # Stop Thread when CTRL + C is pressed or when program is exited
    thread.join()

我尝试过使用 asyncio 的其他解决方案，但无法让其他解决方案发挥作用。

后续：当您创建一个线程时，当您停止程序（即KeyboardInterupt或SystemExit）时，如何停止该线程？

任何帮助将不胜感激，谢谢！

我正在构建一个测试项目来学习 asyncio。我正在构建一个通过代理服务器获取多个页面的程序。

它对于 http 页面工作正常，但对于 https 页面则失败。当我使用常规请求库时，我也可以使用 https 页面，但不能使用 asyncio。

我隔离了损坏的代码：

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url, proxy="http://192.168.0.2:9001") as response:
            print(await response.text())

loop = asyncio.get_event_loop()
loop.run_until_complete(fetch("https://google.com")) #http websites do work

错误：

Traceback (most recent call last):
  File "C:\Users\GG\AppData\Local\Programs\Python\Python38-32\lib\site-packages\aiohttp\connector.py", line 936, in _wrap_create_connection
    return await self._loop.create_connection(*args, **kwargs)  # type: ignore  # noqa
  File "C:\Users\GG\AppData\Local\Programs\Python\Python38-32\lib\asyncio\base_events.py", line 1050, in create_connection
    transport, protocol = await self._create_connection_transport(
  File "C:\Users\GG\AppData\Local\Programs\Python\Python38-32\lib\asyncio\base_events.py", line 1080, in _create_connection_transport
    await waiter
  File …

我正在学习如何以用户身份登录时使用电视马拉松。目前我在创建内联按钮时遇到一些麻烦。我的代码是：

@register(outgoing=True, pattern="^\.button(?: |$)(.*)")
async def buttontest(test):
    await test.client.send_message(test.chat_id, 'click me', buttons=[Button.inline('Test', 'test-return')])

当我使用它时，我只看到一条“单击我”消息，没有按钮。作为用户使用 telethon 时按钮是否可用？

import robloxapi, asyncio
client = robloxapi.Client(".ROBLOSECURITY Cookie Here") # Removed this for security reasons

async def main():
    user = await client.get_self()

    try:
        role = await user.get_role_in_group(1)
    except robloxapi.utils.errors.NotFound:
        role = None

    if role:    
        print(role.name)
    else:
        print("Not in group")

asyncio.run(main())

这段代码正在引发RuntimeError: Event loop is closed，我不知道为什么，

我尝试用这个替换 asyncio.run

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

但它给了我同样的错误

使用Python 3.6.8

async def sleeper():
    time.sleep(2)

async def asyncio_sleeper():
    await asyncio.sleep(2)

await asyncio.wait_for(sleeper(), 1)

await asyncio.wait_for(asyncio_sleeper(), 1)

使用 time.sleep 不会超时，asyncio.sleep 会超时。

我的直觉是，在协程上调用 wait_for 的超时将基于协程花费的时间，而不是基于协程内的各个异步调用。幕后发生了什么导致了这种行为，有没有办法修改行为以符合我的直觉？

我的 ASGI 应用程序将事件发送到curl，然后发送到我的手机。然而，即使服务器正在发送事件，并且标头看起来正确，但在连接关闭之前，我的 Windows 计算机上的 Firefox 和 Chrome 都不会收到事件。

无论我将服务器托管在 WSL、Powershell 终端还是单独的 Linux 机器上，都会发生这种情况。

然而，如果我将服务器托管在 repl.it 上，这些相同的浏览器可以正常工作（请分叉并尝试一下）。

我尝试修改 Windows 防火墙设置，但无济于事。

这是应用程序代码：

import asyncio
import datetime


async def app(scope, receive, send):
    headers = [(b"content-type", b"text/html")]
    if scope["path"] == "/":
        body = (
            "<html>"
            "<body>"
            "</body>"
            "<script>"
            "  let eventSource = new EventSource('/sse');"
            "  eventSource.addEventListener('message', (e) => {"
            "    document.body.innerHTML += e.data + '<br>';"
            "  });"
            "</script>"
            "</html>"
        ).encode()

        await send({"type": "http.response.start", "status": 200, "headers": headers})
        await send({"type": "http.response.body", "body": …

我有以下场景：

1. 我有一个阻塞同步发电机
2. 我有一个非阻塞异步函数

我想运行阻塞生成器（在 a 中执行ThreadPool）和async事件循环上的函数。我该如何实现这一目标？

以下函数仅打印生成器的输出，而不是sleep函数的输出。

谢谢！

from concurrent.futures import ThreadPoolExecutor

import numpy as np
import asyncio
import time


def f():
    while True:
        r = np.random.randint(0, 3)
        time.sleep(r)
        yield r


async def gen():
    loop = asyncio.get_event_loop()
    executor = ThreadPoolExecutor()
    gen = await loop.run_in_executor(executor, f)
    for item in gen:
        print(item)
        print('Inside generator')


async def sleep():
    while True:
        await asyncio.sleep(1)
        print('Inside async sleep')


async def combine():
    await asyncio.gather(sleep(), gen())


def main():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(combine()) …