相关疑难解决方法(0)

致力于ThreadPoolExecutors使用asyncio和替换所有异步调用所使用的服务器查询工具的实现aiohttp。由于网络调用是非阻塞IO，因此大多数过渡都是直接的，保存响应使我感到困惑。

我正在使用的所有示例，甚至是两个库的文档，都使用该示例asyncio.gather()来收集所有等待的结果。就我而言，这些结果可以是许多GB范围内的文件，我不想将它们存储在内存中。

什么是解决此问题的合适方法？是否可以使用asyncio.as_completed()然后：

for f in as_completed(aws):
    earliest_result = await f
    # Assumes `loop` defined under `if __name__` block outside coroutine
    loop = get_event_loop()
    # Run the blocking IO in an exectuor and write to file
    _ = await loop.run_in_executor(None, save_result, earliest_result)

这不是引入线程（假设ThreadPoolExecutor默认情况下使用a ），从而使它成为异步多线程程序，而不是异步单线程程序吗？

此外，这是否可以确保earliest_result在任何时候仅将1 写入文件？我不想await loop.run_in_executor(...)运行该调用，然后出现另一个结果，我尝试运行到同一文件；我想可以限制一个信号量。

大家好，我在尝试理解asyncio和aiohttp并使两者正常工作方面遇到困难。不仅我不正确地了解自己在做什么，这时我遇到了一个我不知道如何解决的问题。

我正在使用Windows 10 64位最新更新。

以下代码使用asyncio返回了标题中Content-Type中不包含html的页面列表。

import asyncio
import aiohttp

MAXitems = 30

async def getHeaders(url, session, sema):
    async with session:
        async with sema:
            try:
                async with session.head(url) as response:
                    try:
                        if "html" in response.headers["Content-Type"]:
                            return url, True
                        else:
                            return url, False
                    except:
                        return url, False
            except:
                return url, False


def checkUrlsWithoutHtml(listOfUrls):
    headersWithoutHtml = set()
    while(len(listOfUrls) != 0):
        blockurls = []
        print(len(listOfUrls))
        items = 0
        for num in range(0, len(listOfUrls)):
            if num < MAXitems:
                blockurls.append(listOfUrls[num - items])
                listOfUrls.remove(listOfUrls[num - items])
                items …

我有一个生成任务（io 绑定任务）的函数：

def get_task():
    while True:
        new_task = _get_task()
        if new_task is not None:
            yield new_task
        else:
            sleep(1)

我正在尝试在 asyncio 中编写一个消费者，该消费者将同时处理最多 10 个任务，完成一项任务，然后将执行新任务。我不确定是否应该使用信号量或者是否有任何类型的 asycio 池执行程序？我开始用线程写一个伪代码：

def run(self)
   while True:
       self.semaphore.acquire() # first acquire, then get task
       t = get_task()
       self.process_task(t)

def process_task(self, task):
   try:
       self.execute_task(task)
       self.mark_as_done(task)
   except:
       self.mark_as_failed(task)
   self.semaphore.release()

有人可以帮助我吗？我不知道把 async/await 关键字放在哪里

客观的：

我正在尝试同时抓取多个网址。我不想同时发出太多请求，因此我使用此解决方案来限制它。

问题：

正在为所有任务发出请求，而不是一次针对有限数量的任务。

精简代码：

async def download_all_product_information():
    # TO LIMIT THE NUMBER OF CONCURRENT REQUESTS
    async def gather_with_concurrency(n, *tasks):
        semaphore = asyncio.Semaphore(n)

        async def sem_task(task):
            async with semaphore:
                return await task

        return await asyncio.gather(*(sem_task(task) for task in tasks))

    # FUNCTION TO ACTUALLY DOWNLOAD INFO
    async def get_product_information(url_to_append):
        url = 'https://www.amazon.com.br' + url_to_append

        print('Product Information - Page ' + str(current_page_number) + ' for category ' + str(
            category_index) + '/' + str(len(all_categories)) + ' in …