Как запускать параллельно выполняющиеся процессы в python c помощью asyncio или других библиотек?

import asyncio


async def f1():
    print("First function...")
    while True:
        await asyncio.sleep(4)


async def f2():
    print("Second function...")
    while True:
        await asyncio.sleep(4)


ioloop = asyncio.get_event_loop()
tasks = [
    ioloop.create_task(f1()),
    ioloop.create_task(f2())
]
ioloop.run_until_complete(asyncio.wait(tasks))
ioloop.close()

в новых версиях питон Python (≥ 3.7) лучше запускать асинхронные функции так:

async def f1():
    print("First function...")
    await asyncio.sleep(4) # Имитирование работы программы
    print(f'{f1.__name__} Done')


async def f2():
    print("Second function...")
    await asyncio.sleep(4) # Имитирование работы программы
    print(f'{f2.__name__} Done')


async def main():
    await asyncio.gather(f1(), f2())


if __name__ == '__main__':
    asyncio.run(main())

pass блокирующая команда, вместо нее лучше использовать asyncio.sleep как имитирование работы функции. Для одновременного выполнения асинхронных программ нужно добавить их в asyncio.gather(func(), func()...) а затем запустить с помощью asyncio.run

Как я понял из слов @vitdev (спасибо ему огромное) с помощью asincio параллельно можно запустить "только неблокирующий код (это ввод-вывод- запись в файл, работа с сетью..)". Поэтому asincio подойдёт не для любых задач.

P.S @asanisimov объяснил как работает Этот модуль. Всетаки asincio может помочь, просто работает по другому принципу. Спасибо большое @asanisimov

Я нашел решение данного вопроса с помощью модуля threading. Например так:

from threading import Thread

def f1():
    print("Starting first function")
    while True: #Этот код работает бесконечно
        print('First code is working...')

def f2():
    print("Starting second function")
    while True: #Этот код тоже работает бесконечно
        print('Second code is working too...')



class MyThread(Thread):
    """A threading example"""
    
    def __init__(self, name, function): # name - некое имя потока (Можно было и без этого обойтись)
                                        # function - функция, которая будет запущена параллельно основному потоку
        """Инициализация потока"""
        Thread.__init__(self)
        self.name = name
        self.function = function
    
    def run(self):
        """Запуск потока"""
        self.function()

    
def create_threads():
    """Создаем группу потоков"""
    
    name = "Thread 1"
    function = f1
    my_thread1 = MyThread(name, function)
    my_thread1.start()

    name = "Thread 2"
    function = f2
    my_thread2 = MyThread(name, function)
    my_thread2.start()
 
 
if __name__ == "__main__":
    create_threads()

Вывод у этой программы довольно страшный (Не буду вставлять), но это значит что все работает.

Можно заменить print('First code is working...') и print('Second code is working too...') на pass и убедиться, что обе функции запустились:

from threading import Thread

def f1():
    print("Starting first function")
    while True: #Этот код работает бесконечно
        pass

def f2():
    print("Starting second function")
    while True: #Этот код тоже работает бесконечно
        pass



class MyThread(Thread):
    """A threading example"""
    
    def __init__(self, name, function): # name - некое имя потока
                                        # function - функция, которая будет запущена параллельно основному потоку
        """Инициализация потока"""
        Thread.__init__(self)
        self.name = name
        self.function = function
    
    def run(self):
        """Запуск потока"""
        self.function()

    
def create_threads():
    """Создаем группу потоков"""
    
    name = "Thread 1"
    function = f1
    my_thread1 = MyThread(name, function)
    my_thread1.start()

    name = "Thread 2"
    function = f2
    my_thread2 = MyThread(name, function)
    my_thread2.start()
 
 
if __name__ == "__main__":
    create_threads()

Вывод:

Starting first function
Starting second function

Возможно я не очень понял суть библиотеки asyncio (Возможно лучше использовать multiprocessing или что-то ещё) Все зависит от вашей задачи, Ваш пример, не работает параллельно, потому что в функциях выполняются блокирующие операции.

Ядро асинхронного приложения - это событийный цикл. В вашем примере, не вдаваясь в детали, ioloop = asyncio.get_event_loop() - вы его получаете, tasks = [ ioloop.create_task(f1()), ioloop.create_task(f2()) ] - создаете задачи для него, ioloop.run_until_complete(asyncio.wait(tasks)) запускаете эти задачи внутри него. Цикл событий выполняет только одну задачу, до тех пор пока не встречает неблокирующую операцию, когда задача Task выполняется с оператором await, то выполняющаяся задача приостанавливается, а цикл обработки событий выполняет следующую задачу. Для вашего примера достаточно добавить asyncio.sleep(0.1) что бы цикл событий мог передавать контекст выполнения между задачами.

import asyncio

async def f1():
    print("First function...")
    while True:  # Этот код работает бесконечно
        await asyncio.sleep(0.1) #-> на этом месте цикл событий переключит выполнение на другую задачу
        print('First code is working...')

async def f2():
    print("Second function...")
    while True:  # Этот код тоже работает бесконечно, но не мешает первому
        await asyncio.sleep(0.1)#-> на этом месте цикл событий переключит выполнение на другую задачу
        print('Second code is working too...')


ioloop = asyncio.get_event_loop()
tasks = [
    ioloop.create_task(f1()),
    ioloop.create_task(f2())
]
ioloop.run_until_complete(asyncio.wait(tasks))
ioloop.close()

Если Вы не используете асинхронные библиотеки в задачах, или другие i/o wait операции ( к примеру открытий файла и чтение из него, тут можно было использовать потоки), а распаралеливание затрагивает только cpu операции, то необходимо использовать процессы multiprocessing. Как пример:

import multiprocessing as mp
def f1():
    print("First function...")
    while True:  # Этот код работает бесконечно
        print('First code is working...')
        time.sleep(0.1)

def f2():
    print("Second function...")
    while True:  # Этот код тоже работает бесконечно, но не мешает первому
        print('Second code is working too...')
        time.sleep(0.1) # это не обязательно

def main(func: callable):
    func()

tasks = [f1, f2]
procs = [mp.Process(target=main, args=(i,)) for i in tasks] # создаем столько процессов, сколько имеем функций
for proc in procs:
    proc.start()

for proc in procs:
    proc.join()

На практике чаще имеется одна функция и много аргументов, и необходимость распаралелить выполнение одной функции. Тогда кол-во аргументов бьется на пачки и передается пачка в каждый процесс, если кол-во процессов < "пачек" то лучше использовать пул процессов (multiprocessing.Pool), так как создание процесса считается тяжелой операцией, а так у Вас заранее будет выделено N процессов в вашем пуле.

Если функция асинхронная

async def f1():
    print("Async function...")
    while True:
        await asyncio.sleep(5)

Если функция синхронная

async def f2():
    print("Blocking function...")
    while True:
        await ioloop.run_in_executor(None, time.sleep(5))

Если на функции не блокирующий опрос состояния можно сделать так - await asyncio.sleep(0) выполняется на каждом шаге лупа пока другой таск висит на слове await в конкурентном таске

async def f2():
    print("Check function...")
    while True:
        await asyncio.sleep(0)
        if driver.state == 1:
            ioloop.call_soon(cb)

await asyncio.sleep(0) можно использовать как переключатель между тасками если не беда что они будут блокироваться - например при какой-нибудь математике, когда ещё не выгодно уходить на мультипроцессинг.

Тут колбэк может снова заблокировать луп - так что смотрим выше