Мультипроцессинг python

Рейтинг: 0Ответов: 1Опубликовано: 05.03.2023

Всем привет. Нужна ваша помощь! У меня есть парсер-бот, который сначала получает данные с 3 источников, потом берет каждый элемент по очереди из LIST, взаимодействует с ним и полученными данными, делает проверку, выводит нужные и потом берет следующий элемент из LIST, когда закончатся элементы - делает снова запрос к 3 источникам и снова проверяет все эти элементы. Мне нужно, чтобы бот только один раз получил данные из requests, и потом в несколько процессов пошел проверять список LIST по этим данным. Как список закончился - снова один раз получил эти данные и пошел проверять список. Я пытался применять пул, но у меня совсем что-то не получилось. Буду очень благодарен за любой совет!

def requests():
    a = session.get...
    b = session.get...
    c = session.get...
def main():
    for item in LIST:
# берет элемент, обрабатывает его и отдает через yield
# этот элемент потом идет через много фильтров, если пройдет - выводится значение, если 
нет - берется следующий элемент, который снова идет через фильтры

While True:
    requests()
    main()

UPD: нашел такое решение, очень простое и идеально для меня подходящее. a, b, c - это данные полученные с requests, которые идут аргументами в main вместе с итерируемым TOKENS.

if __name__ == '__main__':
        session = requests.Session()
        while True:
            requests()
            with multiprocessing.Pool(multiprocessing.cpu_count()) as pool:
                pool.map(partial(main,a,b,c), TOKENS)
python
Источник: Stack Overflow на русском

Ответы

▲ 2Принят

Как-то так:

from concurrent.futures import ThreadPoolExecutor
from queue import Empty, Queue

def requests(tasks: Queue[int]) -> None:
    tasks.put(1)
    tasks.put(2)
    tasks.put(3)
    tasks.put(4)
    print("requests: new tasks added")


def processor(proc_id: int, tasks: Queue[int]) -> None:
    while True:
        try:
            val: int = tasks.get(block=False)
            print(f"processor {proc_id}: processing {val}")
            tasks.task_done()
        except Empty:
            pass


def main() -> None:
    queue: Queue[int] = Queue()
    with ThreadPoolExecutor(max_workers=2) as executor:
        executor.submit(processor, 1, queue)
        executor.submit(processor, 2, queue)
        for i in range(3):
            requests(queue)
            print("main: waiting all processed")
            queue.join()
            print("main: all processed")

main()