Назначение отдельных объектов каждому Python многопроцессорному потоку

Question

У меня есть функция, которую я сопоставляю нескольким рабочим потокам для выполнения. Я хотел бы, чтобы каждый поток поддерживал свой собственный словарь для записи результатов из функции и, наконец, индивидуально записывал содержимое каждого словаря в отдельные файлы. Я хотел бы знать, как выполнить sh это в Python. Я не нашел четкого способа присвоить объект одному потоку. Документация охватывает только информацию, относящуюся к разделению памяти между потоками (с объектами Manager). Ниже приведен код, который я использую для работы (за исключением current_thread_dict в работнике, в настоящее время я использую Manager.dict, который используется каждым потоком):

    from multiprocessing import Pool

    def worker(row):
         #add items to current_thread_dict


    with Pool(processes=16) as p:
         results = p.map(worker, rows, chunksize=1)

Answer 1

В дополнение к строке вы можете передать индекс каждому работнику, чтобы он знал, в каком файле он должен сохранить dict. Я реализовал простой пример следующим образом:

from multiprocessing import Pool
import json


def worker(args):
    i, row = args

    # create object
    current_thread_dict = {
        'row': row,
        'ind': i,
    }

    # add stuff to dict
    if not row.startswith('data'):
        current_thread_dict['special'] = row.upper()
    else:
        current_thread_dict['special'] = None
    current_thread_dict['summary'] = '{row} - {ind} (special: {special})'\
        .format(**current_thread_dict)

    # save dict
    with open(f'output{i}.json', 'w') as f:
        json.dump(current_thread_dict, f, indent=2)


def main():
    rows = ['data1', 'wow', 'data3', 'banana']
    with Pool(processes=16) as p:
        results = p.map(worker, enumerate(rows), chunksize=1)
    print('results:', results)


if __name__ == '__main__':
    main()

Надеюсь, это было полезно!