Класс Manager() модуля multiprocessing в Python

import multiprocessing

proxy = multiprocessing.Manager()

with Manager() as manager:
    # Пустое пространство имен
    shared_ns = manager.Namespace()
    
    # С начальными атрибутами
    shared_ns = manager.Namespace(x=1, y=2, data=[1,2,3])

def worker(ns, value):
    ns.result = value * 2  # Добавляем новый атрибут
    ns.x += 1  # Модифицируем существующий

if __name__ == '__main__':
    with Manager() as manager:
        ns = manager.Namespace(x=0, y=0)
        
        procs = [Process(target=worker, args=(ns, i)) 
                for i in range(5)]
        
        for p in procs:
            p.start()
        for p in procs:
            p.join()
            
        print(f"Результат: {ns.result}, x={ns.x}")
        # Возможный вывод: Результат: 8, x=5

config = manager.Namespace(
    debug=True,
    timeout=30,
    services=manager.list(['auth', 'db'])
)

def worker(conf):
    if conf.debug:
        print(f"Timeout: {conf.timeout}")

stats = manager.Namespace(
    requests=0,
    errors=0,
    processing_time=0.0
)

def update_stats(s, time_taken):
    s.requests += 1
    s.processing_time += time_taken

with Manager() as manager:  # Автоматическое закрытие
    ns = manager.Namespace()
    # работа с пространством имен

from multiprocessing import Manager, Process

def worker(d):
    d['count'] += 1

if __name__ == '__main__':
    with Manager() as manager:
        shared_dict = manager.dict({'count': 0})  # Создаём общий словарь

        print("До:", shared_dict)  # До: {'count': 0}

        # Запускаем несколько процессов
        processes = [Process(target=worker, args=(shared_dict,)) for _ in range(5)]
        for p in processes:
            p.start()
        for p in processes:
            p.join()

        print("После:", shared_dict)  # После: {'count': 5}

with Manager() as manager:
    d = manager.dict()
    d['list'] = manager.list([1, 2, 3])
    d['name'] = 'Alice'

    print(d)  # {'list': [1, 2, 3], 'name': 'Alice'}

    d['list'].append(4)
    print(d['list'])  # [1, 2, 3, 4]

from multiprocessing import Manager

with Manager() as manager:
    # пустой список
    shared_list = manager.list() 
    # или с начальными данными
    shared_list = manager.list([1, 2, 3])

from multiprocessing import Process, Manager

def add_item(lst, value):
    lst.append(value)

if __name__ == '__main__':
    with Manager() as manager:
        data = manager.list()

        # Запускаем несколько процессов
        processes = [Process(target=add_item, args=(data, i)) for i in range(5)]
        for p in processes:
            p.start()
        for p in processes:
            p.join()

        print(data)  # Вывод: [0, 1, 2, 3, 4]

with Manager() as manager:
    # Пустое множество
    shared_set = manager.set()
    
    # С начальными значениями (Python 3.14+)
    shared_set = manager.set({1, 2, 3})  # или manager.set([1, 2, 3])

def process_task(s, item):
    s.add(item ** 2)  # Добавляем квадрат числа
    if len(s) > 5:
        s.clear()  # Очищаем при превышении размера

if __name__ == '__main__':
    with Manager() as manager:
        numbers = manager.set()
        
        # Запускаем процессы
        procs = [Process(target=process_task, args=(numbers, i)) 
                for i in range(10)]
        
        for p in procs:
            p.start()
        for p in procs:
            p.join()
            
        print(f"Итоговый набор: {numbers}")
        # Возможный вывод: {0, 1, 4, 9, 16, 25, 36, 49, 64, 81}

# Поддержка операторов множеств
s1 = manager.set({1, 2})
s2 = manager.set({2, 3})

s1 |= s2       # Объединение (аналог update())
s1 & s2        # Пересечение
s1 - s2        # Разность

import multiprocessing
import time

def worker(ns, lst_mgr, dict_mgr):
    # Изеняем пространство имен, 
    # с ним все нормально
    ns.x += 1
    ns.y[0] += 1
    ns_z = ns.z
    ns_z[0] += 1
    ns.z = ns_z

    # Меняем список
    lst_mgr[0] += 1
    # прямые изменения простого списка 
    # внутри прокси-списка не распространяться 
    # на другие процессы
    lst_mgr[1][0] += 1
    # изменения распространяться на другие
    # потоки только после прямого присвоения 
    # нового списка элементу прокси-списка
    temp_list = lst_mgr[2]
    temp_list[0] += 10
    temp_list.append('100')
    lst_mgr[2] = temp_list
    # а если изменять прокси-список внутри 
    # прокси-списка, то будет все нормально 
    lst_mgr[3][0] += 5
    lst_mgr[3].insert(0, 'вложенный прокси-список')

    # то же самое будет и со словарями
    dict_mgr[0] += 1
    # изменения не увидим
    dict_mgr[1][0] += 1
    # изменяем простой список (значение 2 ключа словаря)
    dict_val_temp = dict_mgr[2]
    dict_val_temp[0] += 10
    dict_val_temp.append('100')
    dict_mgr[2] = dict_val_temp
    dict_mgr[3][0] += 5
    dict_mgr[3].insert(0, 'прокси-список, прокси-словаря')

if __name__ == '__main__':
    manager = multiprocessing.Manager()
    ns_mgr = manager.Namespace()
    ns_mgr.x = 1
    ns_mgr.y = [1]
    ns_mgr.z = [1]
    lst_mgr = manager.list([1, [1], [1], manager.list([1])])
    dict_mgr = manager.dict({0: 1, 1: [1], 2: [1], 3: manager.list([1])})

    print('\nДо запуска процесса:')
    print('-' * 25)
    print('Пространство имен:', ns_mgr)
    print('Прокси-список:', lst_mgr)
    print('lst_mgr[3]:', list(lst_mgr[3]))
    print('Прокси-словарь:', dict_mgr)
    print('dict_mgr[3]:', list(dict_mgr[3]))
    p = multiprocessing.Process(target=worker, args=(ns_mgr, lst_mgr, dict_mgr))
    p.start()
    p.join()
    print('\nПосле запуска процесса:')
    print('-' * 50)
    print('Пространство имен:', ns_mgr)
    print('Прокси-список:', lst_mgr)
    print('lst_mgr[3]:', list(lst_mgr[3]))
    print('Прокси-словарь:', dict_mgr)
    print('dict_mgr[3]:', list(dict_mgr[3]))


# До запуска процесса:
# -------------------------
# Пространство имен: Namespace(x=1, y=[1], z=[1])
# Прокси-список: [1, [1], [1], <ListProxy object, typeid 'list' at 0x7f34cedde040>]
# lst_mgr[3]: [1]
# Прокси-словарь: {0: 1, 1: [1], 2: [1], 3: <ListProxy object, typeid 'list' at 0x7f34cedde3a0>}
# dict_mgr[3]: [1]

# После запуска процесса:
# --------------------------------------------------
# Пространство имен: Namespace(x=2, y=[1], z=[2])
# Прокси-список: [2, [1], [11, '100'], <ListProxy object, typeid 'list' at 0x7f34cedde040>]
# lst_mgr[3]: ['вложенный прокси-список', 6]
# Прокси-словарь: {0: 2, 1: [1], 2: [11, '100'], 3: <ListProxy object, typeid 'list' at 0x7f34cedde3a0>}
# dict_mgr[3]: ['прокси-список, прокси-словаря', 6]

import multiprocessing, time, random

def worker(lst, dkt, lock):
    pid_proc = multiprocessing.current_process().pid
    proc_name = multiprocessing.current_process().name
    # блокируем доступ к прокси-списку из других потоков 
    lock.acquire()
    try:
        for _ in range(3):
            # имитируем нагрузку, для того, что бы была 
            # конкуренция доступа к общему ресурсу (списку)
            time.sleep(random.uniform(0.01, 0.1))
            # добавляем PID процесса 3 раза
            lst.append(pid_proc) 
        
        # если в словаре нет записи о 
        # процессе, то добавляем
        if not dkt.get(pid_proc, None):
            dkt[pid_proc] = proc_name

    finally:
        # завершаем процесс и разрешаем 
        # доступ к массиву другим процессам
        lock.release()

if __name__ == '__main__':
    manager = multiprocessing.Manager()
    # создаем прокси список
    lst = manager.list()
    # создаем прокси словарь
    dkt = manager.dict()
    # создаем прокси блокировщик
    lock = manager.Lock()
    
    procs = []
    for _ in range(3):
        proc = multiprocessing.Process(target=worker, args=(lst, dkt, lock))
        procs.append(proc)
        proc.start()

    # Ждем результатов
    [proc.join() for proc in procs]
    print('Вывод результатов:')
    print(lst)
    print(dkt)
    # очищаем используемые ресурсы
    [proc.close() for proc in procs]
    
    
# Вывод результатов:
# [86138, 86138, 86138, 86139, 86139, 86139, 86140, 86140, 86140]
# {86138: 'Process-2', 86139: 'Process-3', 86140: 'Process-4'}