Как запустить код блокировки независимо от цикла asyncio
Мой проект требует, чтобы я запускал блокирующий код (из другой библиотеки), продолжая при этом цикл asyncio while: true. Код выглядит примерно так:
async def main():
while True:
session_timeout = aiohttp.ClientTimeout()
async with aiohttp.ClientSession() as session:
// Do async stuffs like session.get and so on
# At a certain point, I have a blocking code that I need to execute
// Blocking_code() starts here. The blocking code needs time to get the return value.
Running blocking_code() is the last thing to do in my main() function.
# My objective is to run the blocking code separately.
# Such that whilst the blocking_code() runs, I would like my loop to start from the beginning again,
# and not having to wait until blocking_code() completes and returns.
# In other words, go back to the top of the while loop.
# Separately, the blocking_code() will continue to run independently, which would eventually complete
# and returns. When it returns, nothing in main() will need the return value. Rather the returned
# result continue to be used in blocking_code()
asyncio.run(main())
Я пробовал использовать pool = ThreadPool(processes=1) и thread = pool.apply_async(blocking_code, params). Это как бы работает, если есть вещи, которые нужно сделать после blocking_code() внутри main(); но blocking_code() — это последняя функция в main(), и это приведет к паузе всего цикла while до тех пор, пока не завершится blocking_code(), прежде чем начать сначала.
Я не знаю, возможно ли это, и если да, то как это делается; но идеальный сценарий таков.
Запустите main(), затем запустите blocking_code() в своем собственном экземпляре. Как будто выполняется другой файл .py. Таким образом, как только цикл достигает blocking_code() в main(), он запускает файл blocking_code.py, и пока выполняется скрипт blocking_code.py, циклы while снова продолжаются сверху.
Если к моменту 2-го прохода цикла while он снова достигает blocking_code(), а предыдущий запуск не завершен; другой экземпляр blocking_code() будет работать в своем собственном экземпляре независимо.
Имеет ли смысл то, что я говорю? Возможно ли добиться желаемого результата?
Благодарю вас!
Ответы 1
Это возможно с помощью нитей. Таким образом, вы не блокируете свой основной цикл, вам нужно обернуть поток в задачу asyncio. Вы можете дождаться возвращаемых значений после завершения цикла, если вам это нужно. Вы можете сделать это с помощью комбинации asyncio.create_task и asyncio.to_thread
import aiohttp
import asyncio
import time
def blocking_code():
print('Starting blocking code.')
time.sleep(5)
print('Finished blocking code.')
async def main():
blocking_code_tasks = []
while True:
session_timeout = aiohttp.ClientTimeout()
async with aiohttp.ClientSession() as session:
print('Executing GET.')
result = await session.get('https://www.example.com')
blocking_code_task = asyncio.create_task(asyncio.to_thread(blocking_code))
blocking_code_tasks.append(blocking_code_task)
#do something with blocking_code_tasks, wait for them to finish, extract errors, etc.
asyncio.run(main())
Приведенный выше код запускает блокирующий код в потоке, а затем помещает его в асинхронную задачу. Затем мы добавляем это в список blocking_code_tasks, чтобы отслеживать все текущие задачи. Позже вы можете получить значения или ошибки с помощью чего-то вроде asyncio.gather