使用Multiprocessing Value(ctypes.c_wchar_p)致线程阻塞，求替代方案

我用multiprocessing和threading实现并发，需求是从多进程中创建多线程，线程要访问通过Value()实现的全局变量，实现跨进程（及线程）的变量修改与同步。结构大致如下：

from multiprocessing import Queue, Process, Value
dc = {A: Queue(cpu_count), B: Value(ctypes.c_wchar_p, '', lock = False), C: Value('i', 0, lock = False)}
# Process1作为辅助进程创建工作线程：
# Threads1为Queue生成任务
# Threads2通过dc[c].value传递KeyboardInterrupt标志信号
# Process3-8作为主进程创建多线程：
# Process3-8的Threads1-4访问dc[b].value进行连接

但访问dc[b].value时线程会阻塞，打印dc[b]则正常，输出为：

c_wchar_p(2184070568752)

我写了最小复现示例：

from multiprocessing import Process, Queue, Manager, cpu_count, Value, Lock
import threading
import ctypes
import time

def print_props():
    while True:
        print('Process1 ', props['A'].value)

def change_props():
    while True:
        if props['A'].value == '0':
            props['A'].value = '1'
        else:
            props['A'].value = '0'
        print('Process2 ', props['A'].value)

def props_thread(n):
    global props
    props = n
    task_producer_thread = threading.Thread(target = change_props)
    task_producer_thread.start()
    task_producer_thread.join()

def main_thread(n):
    global props
    props = n
    task_producer_thread = threading.Thread(target = print_props)
    task_producer_thread.start()
    task_producer_thread.join()

def main():
    parent_process = []
    props_thread_process = Process(target = main_thread, args = (props, ))
    props_thread_process.start()
    parent_process.append(props_thread_process)
    
    main_thread_process = Process(target = props_thread, args = (props, ))
    main_thread_process.start()
    parent_process.append(main_thread_process)
    
    for i in parent_process:
        i.join()

if __name__ == '__main__':
    global props
    props = {"A": Value(ctypes.c_wchar_p, '0', lock = False), "B": Value('i', 0, lock = False)}
    main()

可以看到Value(ctypes.c_wchar_p, '', lock = False)无法跨线程和进程正常访问，但Value('i', 0, lock = False)可以正常工作无阻塞。我希望多进程内的线程也能访问Value()，调用dc[b].value时返回字符串，请问如何实现跨进程和线程的可编辑全局字符串变量？

环境：Python 3.12，Windows 11 22H2

问题根源 #

ctypes.c_wchar_p是指针类型，它存储的是字符串在当前进程内存空间的地址。跨进程时，这个地址在其他进程的内存空间中完全无效——要么指向错误的内存区域，要么直接引发阻塞/崩溃，这就是你看到访问value时阻塞的原因。而整数类型的Value（比如'i'）直接存储值本身，不是指针，所以能跨进程正常访问。

可行实现方式 #

1. 使用multiprocessing.Manager管理共享字符串

Manager会创建一个单独的服务进程来管理共享对象，所有进程通过代理访问，天然支持跨进程+跨线程的字符串共享：

from multiprocessing import Process, Manager, Lock
import threading
import time

def print_props():
    while True:
        print('Process1 ', props['A'])
        time.sleep(0.5)

def change_props():
    while True:
        if props['A'] == '0':
            props['A'] = '1'
        else:
            props['A'] = '0'
        print('Process2 ', props['A'])
        time.sleep(0.5)

def props_thread(n):
    global props
    props = n
    task_producer_thread = threading.Thread(target=change_props)
    task_producer_thread.start()
    task_producer_thread.join()

def main_thread(n):
    global props
    props = n
    task_producer_thread = threading.Thread(target=print_props)
    task_producer_thread.start()
    task_producer_thread.join()

def main():
    with Manager() as manager:
        props = manager.dict({"A": '0', "B": 0})
        parent_process = []
        
        props_thread_process = Process(target=main_thread, args=(props,))
        props_thread_process.start()
        parent_process.append(props_thread_process)
        
        main_thread_process = Process(target=props_thread, args=(props,))
        main_thread_process.start()
        parent_process.append(main_thread_process)
        
        for p in parent_process:
            p.join()

if __name__ == '__main__':
    main()

• 注意：Manager的代理对象自带锁，不需要手动加锁就能保证线程和进程安全；如果需要更细粒度的控制，可以手动添加Lock。

2. 使用固定大小的字符数组作为Value的类型

如果不想用Manager，可以用ctypes.c_wchar * N创建固定长度的字符数组，把字符串存到共享内存中：

from multiprocessing import Process, Value, Lock
import threading
import ctypes
import time

# 定义固定长度的字符数组类型，比如最多存100个字符
StringValue = ctypes.c_wchar * 100

def print_props():
    while True:
        # 把数组转换成字符串，去掉末尾的空字符
        current_str = props['A'].value.value.strip('\x00')
        print('Process1 ', current_str)
        time.sleep(0.5)

def change_props():
    while True:
        current_str = props['A'].value.value.strip('\x00')
        new_str = '1' if current_str == '0' else '0'
        # 把新字符串复制到数组中，自动补空字符
        props['A'].value.value = new_str
        print('Process2 ', new_str)
        time.sleep(0.5)

def props_thread(n):
    global props
    props = n
    task_producer_thread = threading.Thread(target=change_props)
    task_producer_thread.start()
    task_producer_thread.join()

def main_thread(n):
    global props
    props = n
    task_producer_thread = threading.Thread(target=print_props)
    task_producer_thread.start()
    task_producer_thread.join()

def main():
    parent_process = []
    # 初始化固定长度字符数组，初始值为'0'
    props = {"A": Value(StringValue, '0', lock=True), "B": Value('i', 0, lock=True)}
    
    props_thread_process = Process(target=main_thread, args=(props,))
    props_thread_process.start()
    parent_process.append(props_thread_process)
    
    main_thread_process = Process(target=props_thread, args=(props,))
    main_thread_process.start()
    parent_process.append(main_thread_process)
    
    for p in parent_process:
        p.join()

if __name__ == '__main__':
    main()

• 注意：这种方式需要提前确定字符串的最大长度；必须启用lock=True（默认就是True），避免多线程/进程同时修改导致数据错乱。

内容的提问来源于stack exchange，提问作者Guild