Python跨进程访问共享字符串出现无错误崩溃，求原因及解决方法

问题原因 #

你用mp.Value(c_wchar_p, '')实现进程间字符串共享的方式存在致命问题：c_wchar_p是宽字符指针类型，mp.Value在这里共享的仅仅是指针的内存地址，而非字符串的实际内容。

子进程生成的字符串存储在自身的私有内存空间中，当你把newString赋值给self.sharedString.value时，实际是把该字符串在子进程内存中的地址写入了共享内存。父进程读取这个指针地址后，会尝试访问不属于自己内存空间的地址，直接触发内存访问错误——Windows系统会直接终止主进程，且不会抛出明显错误提示，同时子进程可能因为主进程异常退出而残留为僵尸进程。

加锁无法解决这个问题，因为锁只能保证读写的原子性，却无法解决跨进程内存地址无效的核心问题。

解决方法 #

以下两种方案可以安全实现进程间字符串共享：

方案1：使用multiprocessing.Array（共享内存方式） #

mp.Array会在共享内存中分配固定大小的字符数组，直接存储字符串内容，避免指针跨进程访问的问题。需要预先设置足够的数组长度来容纳最长的字符串：

import multiprocessing as mp
from ctypes import c_wchar
import time
import random

alphabet = 'abcdefghijklmnopqrstuvwxyz'

class StringGenerator(mp.Process):
    def __init__(self, *args, **kwargs):
        mp.Process.__init__(self, *args, daemon=True, **kwargs)
        # 分配能存储100个宽字符的共享数组（可根据需求调整大小）
        self.sharedString = mp.Array(c_wchar, 100)

    def run(self):
        print('generator init')
        while True:
            newString = ''.join(random.sample(alphabet, int(random.random()*10+5)))
            # 将字符串写入共享数组，自动填充空字符到数组长度
            self.sharedString[:len(newString)] = newString
            print('generator created string:', newString)
            time.sleep(0.5)

if __name__ == '__main__':
    generator = StringGenerator()
    generator.start()

    print('starting')
    while True:
        time.sleep(0.69)
        print('getting string from generator')
        # 从共享数组读取字符串，直到遇到第一个空字符
        newString = ''.join(generator.sharedString).rstrip('\x00')
        print('new string:', newString)
    print('done')

方案2：使用multiprocessing.Manager（IPC通信方式） #

mp.Manager通过中间管理进程实现数据共享，无需担心内存地址问题，用法更灵活，不需要预先设置字符串长度，缺点是性能略低于直接共享内存：

import multiprocessing as mp
import time
import random

alphabet = 'abcdefghijklmnopqrstuvwxyz'

class StringGenerator(mp.Process):
    def __init__(self, shared_string, *args, **kwargs):
        mp.Process.__init__(self, *args, daemon=True, **kwargs)
        self.sharedString = shared_string

    def run(self):
        print('generator init')
        while True:
            newString = ''.join(random.sample(alphabet, int(random.random()*10+5)))
            self.sharedString.value = newString
            print('generator created string:', newString)
            time.sleep(0.5)

if __name__ == '__main__':
    with mp.Manager() as manager:
        # 创建可共享的Unicode字符串值
        shared_string = manager.Value('U', '')
        generator = StringGenerator(shared_string)
        generator.start()

        print('starting')
        while True:
            time.sleep(0.69)
            print('getting string from generator')
            newString = shared_string.value
            print('new string:', newString)
    print('done')

方案对比 #

• 共享内存（Array）：性能更高，适合对速度要求高的场景，但需要预先确定字符串的最大长度。
• Manager方式：使用更灵活，无需限制字符串长度，适合大多数通用场景，性能损失在多数情况下可以忽略。

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