python写共享内存，格式json

Python3

 以下写共享内存，格式为json；

但是需要注意，程序如果退出，写入共享内存数据就失效了！！！

关于锁的问题，根据自己情况添加即可。

_lock = Lock()  声明锁

_lock .acquire()   获得锁

_lock .release()   释放锁

//写共享内存示例

import json
import traceback
from multiprocessing import shared_memory
import time

class BytesJSONEncoder(json.JSONEncoder):
    """用于解决json.dumps函数发现字典里面有bytes类型的数据无法编码问题"""
    def default(self, obj):
        if not isinstance(obj, bytes):
            return json.JSONEncoder.default(self, obj)
        return str(obj, encoding='utf-8')

shared_memory_key = "test_key"

try:
    # 尝试创建共享内存，若失败则映射同名内存空间
    shm = shared_memory.SharedMemory(name=shared_memory_key, create=True, size=1024)  
except:
    shm = shared_memory.SharedMemory(name=shared_memory_key, create=False)

obj = '''
{"a":{ "a1":123,"a2":12,"a3":33}}
'''

try:
    if obj is not None and obj != {}:
        obj_str = json.dumps(obj, cls=BytesJSONEncoder)
        obj_len = len(obj_str)
        content = str(obj_len) + "~" + obj_str
        b = content.encode()
        shm.buf[0:len(b)] = b

        print(f"开始休眠")
        time.sleep(9000000)

        # 关闭共享内存
        shm.close()
        shm.unlink()
        print(f"程序结束!!!")

except Exception as e:
    traceback.print_exc()

//以下为读写共享内存 封装的方法：

def print_from_shm(shm_name):
    with shared_memory.SharedMemory(name=shm_name) as shm:
        # 将共享内存的内容转换为字符串
        # 注意，我们假设共享内存的内容是一个以 null 结尾的字符数组
        string_from_shm = shm.buf.decode('utf-8')
        print(string_from_shm)
 
def write_to_shm(shm_name, data):
    with shared_memory.SharedMemory(name=shm_name, create=True, size=len(data)+1) as shm:
        # 将数据写入共享内存
        shm.buf[0:len(data)] = data.encode('utf-8')
        # 添加 null 字符
        shm.buf[len(data)] = 0