[Python] Shared Memory

writer.py

import numpy as np
import time
from multiprocessing import shared_memory

def writer():
    # 공유 메모리 생성 (size는 데이터 크기에 맞게 설정)
    shm = shared_memory.SharedMemory(create=True, size=4096)

    # 공유 메모리에 연결된 NumPy 배열 생성
    data = np.ndarray((5,), dtype=np.float64, buffer=shm.buf)
    
    print(f"Shared memory name: {shm.name}")
    try:
        # 실시간으로 데이터를 업데이트하는 루프
        for i in range(100):
            data[:] = np.random.random(5)  # 임의의 데이터로 채우기
            print(f"Writer: Updated data {data.round(4)}")
            time.sleep(1)  # 1초마다 데이터 갱신
    finally:
        # 공유 메모리 해제
        shm.close()
        shm.unlink()

if __name__ == "__main__":
    writer()

reader.py

import numpy as np
import time
from multiprocessing import shared_memory

def reader():
    # Writer에서 생성된 공유 메모리 연결
    shm = shared_memory.SharedMemory(name='psm_26749de3')  # 생성된 공유 메모리 이름 사용

    # 공유 메모리에 연결된 NumPy 배열 생성
    data = np.ndarray((256,), dtype=np.float64, buffer=shm.buf)

    try:
        # 실시간으로 데이터를 읽는 루프
        for i in range(100):
            print(f"Reader: Read data {i} - {data[:5].round(4)}")  # 데이터의 앞부분만 출력
            time.sleep(1)  # 1초마다 데이터 읽기
    finally:
        # 공유 메모리 해제
        shm.close()

if __name__ == "__main__":
    reader()

결과 :
Writer: Updated data [0.7114 0.2909 0.8442 0.1114 0.6553]
Reader: Read data 99 - [0.7114 0.2909 0.8442 0.1114 0.6553]