1.各个平台下的共享内存的使用方法

共享内存主要用来进行模块分割，复杂性隔离的用途。python这类跨平台平台其实是无法完全消弭各个平台间的个体差异，一个简单的例子：

import posix_ipc
import mmap# 创建或打开共享内存
shm = posix_ipc.SharedMemory('/example', posix_ipc.O_CREAT, mode=0o666, size=100)# 映射共享内存到当前进程
shmmap = mmap.mmap(shm.fd, shm.size)# 在共享内存中写入数据
shmmap.write(b'Hello, shared memory!')# 在另一个进程中读取共享内存中的数据
shm2 = posix_ipc.SharedMemory('/example')
shmmap2 = mmap.mmap(shm2.fd, shm2.size)content = shmmap2.read()
content = content.rstrip(b'\x00');
print("%s" %content)  # 输出 b'Hello, shared memory!'# 释放共享内存
shmmap2.close()
shm2.close_fd()
shmmap.close()
shm.close_fd()
shm.unlink()

上面是linux下的共享内存使用一例，有多种途径，这是其中的一种。但是windows平台下，它是这样的：

from multiprocessing.shared_memory import SharedMemory
from typing import Iterator
from multiprocessing.managers import SharedMemoryManager
from multiprocessing import Process, Queue
import numpy as np
import uuid
import traceback
import time
from contextlib import contextmanager
import uuid#from process_call_func_test import child_process# 创建子进程，共享内存会被子进程继承
def child_process(name):print("child:",name)try:# 在子进程中访问共享内存数据shm = gp_shared_memory_open(name)shared_data = gp_shared_memory_get_uint8_of(shm)print(shared_data)shared_data = Noneshm.close()except Exception as e:traceback.print_exc()print(f"Exception in child process: {e}")print("child quit!")return 1976def gp_shared_memory_create(name, size):# 自行指定共享内存的名称shm_name = nameshm_size = sizeshm = SharedMemory(name=shm_name, create=True, size=shm_size)return shmdef gp_shared_memory_open(name):# 在子进程中访问共享内存数据shm_name = nameshm = SharedMemory(name=shm_name, create=False)return shmdef gp_shared_memory_get_uint8_of(shm) -> Iterator[SharedMemory]:# 在主进程中写入数据到共享内存ar = np.frombuffer(shm.buf, dtype=np.uint8);return ardef gp_shared_memory_close(shm):shm.close()def gp_get_a_randome_name():# 生成一个新的 GUIDnew_guid = uuid.uuid4()name = str(new_guid)return name# 创建子进程，共享内存会被子进程继承
def test_gp_shared_memory():if __name__ == "__main__":name = gp_get_a_randome_name()shm = gp_shared_memory_create(name, 16)buf_in_u8 = gp_shared_memory_get_uint8_of(shm)buf_in_u8[1] = 19buf_in_u8[2] = 76buf_in_u8 = None# 在子进程中访问共享内存数据# 启动子进程p = Process(target=child_process, args=(shm.name,))p.start()p.join(3.5)if p.is_alive():# 子进程尚未结束，进行相应的处理print("Child Process timeout...")# 可以尝试终止子进程p.terminate()# 等待子进程终止p.join()else:time.sleep(1.5)print("child proc done!")gp_shared_memory_close(shm)if __name__ == "__main__":test_gp_shared_memory()

Windows下面的multiprocessing库根本没有shardmemory这个子项， 它被放置在别的地方。

2.初步结论

Python乃至任何一个组织，都没有足够的能力把所有的差异，或大或小的沟渠填平，你只能适应。分歧是必然的，同步，和谐有，更可能没有。一些思维僵化的人无法在舒适圈外有效拓展认知。今天一清早被朋友数落了一句：你是错把风口当本事。。。

这个地方我卡了大概三个小时，老顽固们大部分工作时间，都耗在卡顿上。不是不行，是脑袋跟不上时代——其实就是不行。调试，代码实验，是小白型程序员的看家本领。

附录A 完整的linux下的shared memory封装

附带有测试例程(Usage)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import traceback
import uuid
import posix_ipc
import mmap
import time
from multiprocessing import Processdef _gp_shared_memory_getmmap_of(shm):# 在主进程中写入数据到共享内存mmap_ret = mmap.mmap(shm.fd, shm.size);return mmap_retdef _gp_shared_memory_create(r_name, r_size):# 自行指定共享内存的名称shm_name = r_nameshm = posix_ipc.SharedMemory(shm_name, posix_ipc.O_CREAT, mode=0o666, size=r_size)mmap = _gp_shared_memory_getmmap_of(shm)return (shm, mmap)def _gp_shared_memory_open(name):# 自行指定共享内存的名称shm_name = nameshm = posix_ipc.SharedMemory(shm_name)mmap = _gp_shared_memory_getmmap_of(shm)return (shm, mmap)def gp_get_a_random_name():# 生成一个新的 GUIDnew_guid = uuid.uuid4()name = str(new_guid)return nameclass gp_shared_memory_owner:def __init__(self, r_name, r_size):if(r_name is None):r_name = gp_get_a_random_name()self._name = r_name(self._shm, self._mmap) = _gp_shared_memory_create(r_name, r_size)self._size = self._shm.sizedef getmmap(self):return self._mmapdef getname(self):return self._namedef getsize(self):return self._sizedef release(self):self._mmap.close()self._shm.close_fd()self._shm.unlink()self._name = Noneself._size = 0self._mmap = Noneself._shm = Noneclass gp_shared_memory_client:def __init__(self, r_name):self._name = r_name(self._shm, self._mmap) = _gp_shared_memory_open(r_name)self._size = self._shm.sizedef getmmap(self):return self._mmap;def getname(self):return self._name;def getsize(self):return self._size;def release(self):self._mmap.close()self._shm.close_fd()self._name = Noneself._size = 0self._mmap = Noneself._shm = None# 创建子进程，共享内存会被子进程继承
def test_child_process(name):print("child:",name)try:# 在子进程中访问共享内存数据shmClient = gp_shared_memory_client(name)content = shmClient.getmmap().read()content = content.rstrip(b'\x00');print("child process got msg: ", content)shmClient.release()except Exception as e:traceback.print_exc()print(f"Exception in child process: {e}")print("child quit!")return 1976# 创建子进程，共享内存会被子进程继承
def test_gp_shared_memory():if __name__ == "__main__":shmOwner1 = gp_shared_memory_owner(None,16)shmOwner1.getmmap().write(b'0x190x76')# 在子进程中访问共享内存数据# 启动子进程p = Process(target=test_child_process, args=(shmOwner1.getname(),))p.start()p.join(3.5)if p.is_alive():# 子进程尚未结束，进行相应的处理print("Child Process timeout...")# 可以尝试终止子进程p.terminate()# 等待子进程终止p.join()else:time.sleep(1.5)print("child proc done!")shmOwner1.release()if __name__ == "__main__":test_gp_shared_memory()