преобразование кода из многопоточного в многопроцессорный

Question

Есть код, который сейчас проходится по папкам и анализирует информацию. Он работает в многопоточном стиле. Мне нужно его в многопроцессорном сделать. И с этим проблема. Не пойму, где именно в коде это реализовать. Метод open открывает файлы по ссылке и проходится построчно по ним, метод calculate рассчитывает по формуле значения из файлов. Мне в функции майн нужно сделать цикл while True и ловить ошибки пустоту, я попробовал сделать с условием if not empty, не не пойму что внутрь поместить. Также где-то нужно применить метод put. Я на всякий случай закомментировал уже сделанные изменения, чтобы было проще понять как код работал в многопоточном стиле. Помогите разобраться с этой задачей:

from utilites import show_result, generate_filenames

#import threading
import multiprocessing


#class Ticker(threading.Thread):
class Ticker(multiprocessing.process):

    def __init__(self, ticket_folder, collector, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.ticket_folder = ticket_folder
        self.name_ticket = ''
        self.volatility = 0
        self.collector = collector#?


    def run(self):
        self.calculate(self.open())

    def open(self):
        price_scope = []
        with open(self.ticket_folder, mode='r') as open_ticker:
            for element in open_ticker:
                scattered_element = element.split(',')
                self.name_ticket = scattered_element[0]
                if scattered_element[2] != 'PRICE':
                    price_scope.append(float(scattered_element[2]))
            return price_scope

    def calculate(self, unsorted):
        unsorted.sort()
        half_sum = (unsorted[0] + unsorted[-1]) / 2
        self.volatility = ((unsorted[-1] - unsorted[0]) / half_sum) * 100


def main(folder):
    collector = multiprocessing.Queue()#?
    zero_tickers = []
    value_key = {}
    sorted_place = []
    tickers = []

    for last_folder in generate_filenames(folder):
        tickers.append(Ticker(last_folder))

    for ticker in tickers:
        ticker.start()
    for ticker in tickers:
        ticker.join()
    while not collector.empty():#?
      pass# 


    for ran_ticker in tickers:
        if ran_ticker.volatility == 0:
            zero_tickers.append(ran_ticker.name_ticket)
        else:
            value_key[ran_ticker.volatility] = ran_ticker.name_ticket
            sorted_place.append(ran_ticker.volatility)
            sorted_place.sort()

    show_result(sorted_place, value_key, zero_tickers)


path = "trades/"
if __name__ == '__main__':
    main(folder=path)

Answer 1

map для этой задачи лучше подходит чем процесс. Тут я протащил name_ticket через ретурн чтоб не усложнять с классом. Лучше через замыкание, но лень

from utilites import show_result, generate_filenames
import multiprocessing

def run(ticket_folder):
    return self.calculate(*self.open(ticket_folder))

def open(ticket_folder):
    price_scope = []
    with open(self.ticket_folder, mode='r') as open_ticker:
        for element in open_ticker:
            scattered_element = element.split(',')
            name_ticket = scattered_element[0]
            if scattered_element[2] != 'PRICE':
                price_scope.append(float(scattered_element[2]))
        return price_scope, name_ticket

def calculate(unsorted, name_ticket):
    unsorted.sort()
    half_sum = (unsorted[0] + unsorted[-1]) / 2
    return ((unsorted[-1] - unsorted[0]) / half_sum) * 100, name_ticket


def main(folder):
    zero_tickers = []
    value_key = {}
    sorted_place = []
    tickers = []

    with multiprocessing.Pool() as pool:
        for volatility, name_ticket in pool.imap_unordered(run, generate_filenames(folder)):
        if volatility == 0:
            zero_tickers.append(name_ticket)
        else:
            value_key[volatility] = name_ticket
            sorted_place.append(volatility)
            sorted_place.sort()

    show_result(sorted_place, value_key, zero_tickers)


path = "trades/"
if __name__ == '__main__':
    main(folder=path)

Не проверял, так что если где чтото не так написал: пишите - поправим

БЛОГ НА HUSL

преобразование кода из многопоточного в многопроцессорный

Ответы (1 шт):