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