如何用Python multiprocessing的Array('c')在进程间共享字符串数据

你之前的代码无法运行，核心问题是对Array('c')的理解有误：它创建的是固定长度的字节数组，每个元素是单个字节，不能直接赋值字符串。下面是正确的实现方案，包含多字符串共享、同步处理等关键细节。

错误原因分析 #

• Array('c')本质是基于共享内存的字节数组，仅支持字节级操作，不能直接存储字符串
• 尝试将字符串直接赋值给数组的单个元素（array[count.value - 1] = "string:"+str(count.value)），会导致类型不匹配错误
• 未规划内存空间：单个字符串需要占用多个字节，必须提前分配足够的共享内存

正确实现代码（多字符串共享） #

下面的代码解决了多进程写入字符串数组的问题，同时处理了进程同步避免数据竞争：

from multiprocessing import Value, Array, Process, Lock
import time

def process1(count, str_array, lock):
    for i in range(5):
        time.sleep(0.3)
        # 加锁确保count修改和数组写入是原子操作
        with lock:
            count.value += 1
            # 每个字符串分配10字节空间，计算起始位置
            start_idx = (count.value - 1) * 10
            # 字符串编码为字节，写入指定区间
            current_str = f"string:{count.value}".encode('utf-8')
            str_array[start_idx:start_idx+len(current_str)] = current_str
            # 剩余位置填充空字节，避免读取时残留数据
            str_array[start_idx+len(current_str):start_idx+10] = b'\x00' * (10 - len(current_str))
        print(f"process1: {count.value}")
        
def process2(count, str_array, lock):
    for i in range(5):
        time.sleep(0.7)
        with lock:
            count.value += 2
            start_idx = (count.value - 1) * 10
            current_str = f"string:{count.value}".encode('utf-8')
            str_array[start_idx:start_idx+len(current_str)] = current_str
            str_array[start_idx+len(current_str):start_idx+10] = b'\x00' * (10 - len(current_str))
        print(f"process2: {count.value}")

if __name__ == '__main__':
    count = Value('i', 0)
    # 15个字符串，每个占10字节，总内存150字节
    str_array = Array('c', 150)
    # 锁用于同步进程对共享资源的访问
    lock = Lock()
    
    p1 = Process(target=process1, args=(count, str_array, lock), daemon=True)
    p2 = Process(target=process2, args=(count, str_array, lock), daemon=True)
    
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()
    
    # 读取并解码所有有效字符串
    result = []
    for i in range(15):
        start = i * 10
        end = start + 10
        # 提取字节直到第一个空字节，解码为字符串
        byte_data = str_array[start:end]
        str_data = byte_data.split(b'\x00')[0].decode('utf-8')
        if str_data:
            result.append(str_data)
    print("共享的字符串列表:", result)
    print("进程结束")

关键细节说明 #

1. 内存规划：根据字符串的最大长度，给每个字符串分配固定字节空间（示例中为10字节），总内存=字符串数量×单字符串字节数
2. 编码解码：写入时将字符串编码为UTF-8字节，读取时提取字节并解码为字符串，用空字节\x00作为字符串结束标记
3. 进程同步：使用Lock确保多个进程修改共享变量（count）和写入数组时不会出现数据竞争，避免值混乱
4. 空字节填充：写入后填充空字节，避免后续读取时混入其他字符串的残留数据

单个字符串共享的简化实现 #

如果只需要共享单个字符串，代码可以更简洁：

from multiprocessing import Array, Process, Lock
import time

def update_shared_str(shared_str, lock):
    time.sleep(1)
    with lock:
        new_str = "Hello from sub-process!".encode('utf-8')
        # 写入字节，剩余位置补空字节
        shared_str[:len(new_str)] = new_str
        shared_str[len(new_str):] = b'\x00' * (len(shared_str) - len(new_str))

if __name__ == '__main__':
    # 分配30字节足够容纳目标字符串
    shared_str = Array('c', 30)
    lock = Lock()
    
    p = Process(target=update_shared_str, args=(shared_str, lock))
    p.start()
    p.join()
    
    # 读取并解码
    result = shared_str[:].split(b'\x00')[0].decode('utf-8')
    print("共享的字符串:", result)

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