Совместное использование объекта пользовательского класса через диспетчер многопроцессорности - PullRequest
Новая
       51

Совместное использование объекта пользовательского класса через диспетчер многопроцессорности

1 голос
/ 06 марта 2020

Я работаю над проектом, в котором у меня есть очень специфичные c журналы, и чтобы эти журналы работали, я создал класс, который в основном содержит словарь, в котором записываются время и действия, предпринятые для каждого шага выполнения. Который позже используется для нескольких аналитических целей. позже я создаю файл JSON из этого словаря и загружаю его в корзину s3.

Все работало нормально, когда у меня был только один основной процесс. Но для эффективности я включил многопроцессорность для конкретной задачи c. Я генерирую 4 процесса, которые вызывают одну и ту же функцию, и я хочу, чтобы объект моего класса, который поддерживает журналы, был общим для всех этих процессов.

Я перебрал другие вопросы StackOverflow, но ни один из них не помог. Пожалуйста, помогите мне, если вы считаете, что это где-то лучший ответ, и я, возможно, пропустил его. 

class JsonLogs:
    def __init__(self, date, site_name, user_uuid, is_login=False, request_data=None):
        self.date = date
        self.attribute = value
        if is_login:
            ...
        else:
            ...


    def add_a_logs(self, process, message):
        self.log_dict['key']['some_key'].append(
            {
             'start_time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'),
             'start_message': message,
             'end_message': None,
             'end_time': None,
             'duration': None

            }
        )

    def update_a_logs(self, process=None, message=None, response_data=None):
        log_obj = next((log for log in self.log_dict["key"]['some_key'] if log['process'] == process),
                       None)

        if response_data:
            self.log_dict['key']['some_key'] = response_data

        if log_obj:
            log_obj['end_time'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')
            log_obj['end_message'] = message
            td = datetime.datetime.now() - datetime.datetime.strptime(log_obj['start_time'], '%Y-%m-%d %H:%M:%S.%f')

            log_obj['duration'] = "{}".format(td.seconds)

    .....
    .....

Есть и другие методы, подобные этому, и затем этот словарь преобразуется в JSON и загружается в корзину s3. , Это не имеет ничего общего с модулем ведения журнала python, поскольку у меня очень специфическое c требование.

Мне нужно, чтобы объект этого класса был разделен с 4 создаваемыми мной процессами.

РЕДАКТИРОВАТЬ

Чтобы кому-то было легче понять, что я действительно пытаюсь достичь, можно понять с помощью двух приведенных ниже примеров.

Один процесс: Это работает 

from multiprocessing import Process


class A:
    def __init__(self, a):
        self.a = a

def a():
    x = A(10)
    b(x)
    print(x.a)

def b(y):
    y.a = 20

a()

ВЫХОД: 20

Мультипроцессинг: что я пытаюсь достичь 

from multiprocessing import Process


class A:
    def __init__(self, a):
        self.a = a


def a():
    x = A(10)
    p1 = Process(target=b, args=(x,))
    p2 = Process(target=b, args=(x,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    # b(x)
    print(x.a)

def b(y):
    y.a = 20


a()

Текущий выход: 10

Что мне нужно: 20

1 Ответ

0 голосов
/ 06 марта 2020

Вам необходимо обмениваться объектами между процессами, используя общую память, поскольку каждый процесс имеет свое собственное пространство памяти. Вы можете поделиться объектами, используя объект управления . Но лучше избегать совместного использования сложных объектов, как указано в инструкции .

. В качестве альтернативы вы можете использовать объект queue . Вот игрушечный пример: 

import datetime
from multiprocessing import Process, Queue

def worker(q):
    for _ in range(5):
        q.put(
        {
            'start_time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'),
            'start_message': 'message'
        })
    q.put(None) # End of the queue

def a():
    q = Queue()
    p1 = Process(target=worker, args=(q,))
    p2 = Process(target=worker, args=(q,))
    p1.start()
    p2.start()

    worker_end = 0
    while True:
        i = q.get()
        if i is None:
            worker_end += 1
        else:
            print(i)

        if worker_end == 2:
            break

    p1.join()
    p2.join()

a()
...