Asyncio 运行第二次使用不同输入的相同任务也关闭了第一个任务

Question

我有一个脚本，其中有多个异步函数，我运行将它们循环使用。一切都运行没问题，除了一项任务，我需要使用不同的输入参数运行两次。

def run(self):
    checks_to_run = self.returnChecksBasedOnInputs()
    loop = asyncio.new_event_loop().run_until_complete(self.run_all_checks_async(checks_to_run))
    asyncio.set_event_loop(loop)
    return self.output


async def run_all_checks_async(self,checks_to_run):
    async with aiohttp.ClientSession() as session:
        check_results = []
        for single_check in checks_to_run:
            if single_check == "cvim_check_storage":    #can run parallel in separate thread for each az
                total_number_of_azs = len(Constants.cvim_azs)+1
                self.log.info(total_number_of_azs)
                for x in range(1,total_number_of_azs):
                    task = asyncio.ensure_future(getattr(self, single_check)(session,x))
            else:
                task = asyncio.ensure_future(getattr(self, single_check)(session))
            check_results.append(task)
        await asyncio.gather(*check_results, return_exceptions=False)

class apiCaller:
    def __init__(self):
        pass
        
    async def callAndReturnJson(self, method, url, headers, session, payload, log):
        sslcontext = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
        try:
            async with session.request(method, url, data=payload, headers=headers,ssl=sslcontext) as response:
                response = await response.json()
                print(str(response))
                return response
        except Exception as e:
            print("here exception")
            raise Exception(str(e))

问题出在这个函数中 - 我运行将它设置了两次，但我注意到当任务的第二个版本转到 return 语句时，第一个任务也会立即关闭。我怎样才能避免这种情况并等到其他任务也完成？

async def cvim_check_storage(self,session, aznumber):
        response = await apiCaller().callAndReturnJson("POST",f"https://{single_cvim_az}/v1/diskmgmt/check_disks",getattr(Constants,cvim_az_headers),session=session, payload=payload,log=self.log)
        self.log.info(str(response))
        self.log.info(str(response.keys()))
        if "diskmgmt_request" not in response.keys():
            self.output.cvim_checks.cvim_raid_checks.results[az_plus_number].overall_status = "FAILED"
            self.output.cvim_checks.cvim_raid_checks.results[az_plus_number].details = str(response)
        return 

        ...rest of the code if above if statement is false

Answer 1

问题在于您如何跟踪任务。您正在使用 task 将新任务添加到 check_results，但在您的一个分支中，您在 for 循环内分配 task。但是，在循环完成之前，您不会将 task 添加到 check_results，因此只会添加最后一个任务。 gather 在完成之前不会等待创建的任何其他任务。

解决方法是在内循环的每次迭代中添加task。有几种不同的拼写方式。

一个选项是在您当前分配给 task 的任何地方调用 check_results.append。

if single_check == "cvim_check_storage":    #can run parallel in separate thread for each az
    total_number_of_azs = len(Constants.cvim_azs)+1
    self.log.info(total_number_of_azs)
    for x in range(1,total_number_of_azs):
        check_results.append(
            asyncio.ensure_future(getattr(self, single_check)(session,x))
        )
else:
    check_results.append(
        asyncio.ensure_future(getattr(self, single_check)(session))
    )

不过，我会更进一步，在创建多个任务时使用列表理解。

if single_check == "cvim_check_storage":    #can run parallel in separate thread for each az
    total_number_of_azs = len(Constants.cvim_azs)+1
    self.log.info(total_number_of_azs)
    check_results.extend(
        [
            asyncio.ensure_future(getattr(self, single_check)(session,x))
            for x in range(1,total_number_of_azs)
        ]
    )
else:
    task = asyncio.ensure_future(getattr(self, single_check)(session))
    check_results.append(task)

Asyncio 运行第二次使用不同输入的相同任务也关闭了第一个任务

Asyncio running same task second time with different input shuts down first task too

python

python-asyncio

Asyncio 运行 第二次使用不同输入的相同任务也关闭了第一个任务

Asyncio running same task second time with different input shuts down first task too

python

python-asyncio

Asyncio 运行第二次使用不同输入的相同任务也关闭了第一个任务