标签: python-asyncio

我编写的程序循环遍历一个范围并找到素数和回文数。作为学习 asyncio 的一部分，我尝试使用 async 重新构建它。但结果并不好。这里异步代码比同步代码花费的时间要长得多。

同步代码

import math
import time


def prime(n):
    limit=int(math.sqrt(n))
    for j in range(2,limit):
        if(n%j==0):
            return 0
    return 1


def pallindrome(n):
    n=str(n)
    m=n[::-1]
    if(m==n):
        return 1
    return  0


a, b, c = 999999999, 9999999, 0
start = time.time()

for i in range(a, b, -1): 
    if(pallindrome(i)):  
        if(prime(i)):
            c+=1
            print(i)
    if(c==20):
        break
print("took --> ", time.time()-start)

结果 ：

999727999
999686999
999676999
999565999
999454999
999434999
999272999
999212999
999070999
998979899
998939899
998898899
998757899
998666899
998565899
998333899
998282899
998202899
998171899
998121899
took …

我使用这个答案中的代码，但是asyncio.exceptions.CancelledError当队列为空时得到。在实际项目中，我将任务添加到消费者的队列中，这就是我使用while True语句的原因

我压缩该代码以使调试更容易：

import asyncio
import traceback


async def consumer(queue: asyncio.Queue):
    try:
        while True:
            number = await queue.get()  # here is exception
            queue.task_done()
            print(f'consumed {number}')
    except BaseException:
        traceback.print_exc()


async def main():
    queue = asyncio.Queue()
    for i in range(3):
        await queue.put(i)
    consumers = [asyncio.create_task(consumer(queue)) for _ in range(1)]
    await queue.join()
    for c in consumers:
        c.cancel()


asyncio.run(main())

和错误：

consumed 0
consumed 1
consumed 2
Traceback (most recent call last):
  File "/Users/abionics/Downloads/BaseAsyncScraper/ttt.py", line 8, in consumer
    number …

在新的 asyncio 框架中，如何编写subprocess.check_output的嵌入式异步等效项来捕获流程执行的完整输出？

是否有书籍或教程显示了如何正确使用asyncio的协议？Web上的所有示例都将IO混合到协议定义中！

我想编写一个解析器，该解析器进行帧解码并将消息转换为python数据结构。解析完此数据结构后，我想将其传递给client。

[       ]-->[*protocol parser*]-->[high level api]-->[           ]
[network]                                            [client code]
[       ]<--[*protocol parser*]<--[high level api]<--[           ]

相应地，高级API的客户端传递python数据结构，高级API将该数据结构传递给我的协议，该协议将其转换为正确的字节/文本表示形式并将其传递给传输层。

我假设这是首先抽象出Protocol类的目的。我不想从协议中响应连接的另一端，但这是大多数Web教程显示的内容！

此外，我想了解python世界中提供了哪个高级接口，它是回调，流接口还是其他？

我一直在为这个问题抓狂

有问题的代码是此处开源项目的一部分：aiosmtpd（我的实际 FOSS 项目的分支，此处）

有问题的文件是这个：main.py

出现问题的代码位于main.py的第139行

这是一个片段：

...
from aiosmtpd.smtp import DATA_SIZE_DEFAULT, SMTP, __version__
...

...

    # args is the result of ArgumentParser.parse_args

    factory = partial(
        SMTP, args.handler,
        data_size_limit=args.size, enable_SMTPUTF8=args.smtputf8)

    ...

    server = loop.run_until_complete(
        loop.create_server(factory, host=args.host, port=args.port))

    ...

有时- 也就是说，并非总是- 代码在此时失败RuntimeError: Event loop stopped before Future completed.

我的问题：

1. 这种间歇性故障的原因可能是什么？

   在测试过程中，大约有 10% 的时间会失败（使用 tox + nostest2），但 90% 的时间都进展顺利。

2. 我应该如何检测和/或检查和/或断言以防止这种情况发生？

3. “恢复”错误并重做操作的最佳策略是什么？

我将第n次强调错误是间歇性发生的。虽然不常见，但经常发生，所以我觉得有必要追查根本原因。发生错误后，如果我立即或延迟后再次重新运行代码，则几乎总是不会发生相同的错误。

我最近开始玩 discord bots，我有一个疑问，即是否必须async对不和谐命令使用函数。如果是，那么有人可以告诉我区别吗（根据不和谐的机器人行为）

import time
import asyncio
import aiohttp

async def is_name_available(s, name):
    async with s.get("https://twitter.com/%s" % name) as res:
        if res.raise_for_status == 404:
            print('%s is available!' % name)
            return name

async def check_all_names(names):
    async with aiohttp.ClientSession(raise_for_status=True) as s:
        tasks = []
        for name in names:
            task = asyncio.create_task(is_name_available(s, name))
            tasks.append(task)
        return await asyncio.gather(*tasks)

def main():    
    with open('names.txt') as in_file, open('available.txt', 'w') as out_file:        
        names = [name.strip() for name in in_file]
        start_time = time.time()
        results = asyncio.get_event_loop().run_until_complete(check_all_names(names))
        results = [i for i in …