### Python 多进程共享时间类精准计时 为了实现在多个进程中共享同一个时间源并保持高精度的时间同步，可以采用以下几种方式： #### 使用 `multiprocessing.Manager` 和 `datetime` 通过 `Manager()` 创建的对象可以在不同进程间共享数据。这里展示如何创建一个管理器来维护统一的时间对象。 ```python from multiprocessing import Process, Manager import time from datetime import datetime def update_time(shared_dict): while True: shared_dict['current_time'] = str(datetime.now()) time.sleep(0.1) if __name__ == '__main__': manager = Manager() shared_data = manager.dict() p = Process(target=update_time, args=(shared_data,)) p.start() try: for _ in range(5): # 打印五次当前时间作为示例 print(f"Current Time from process: {shared_data.get('current_time')}") time.sleep(1) finally: p.terminate() # 清理子进程资源 ``` 这种方法简单易懂，但是由于每次访问都需要序列化/反序列化的开销，在追求极高频率更新场景下可能不是最优解[^1]。 #### 利用 POSIX 共享内存 (Linux/macOS 特定功能) 如果操作系统支持 POSIX 共享内存，则可以通过它更高效地实现跨进程的数据交换。下面是一个基于此特性的例子: ```python import posix_ipc as ipc import mmap import struct import os import ctypes import numpy as np from multiprocessing import Process class SharedTime(ctypes.Structure): _fields_ = [("timestamp", ctypes.c_double)] def write_to_shared_memory(shm_name="/my_shm"): memory = None try: semaphore = ipc.Semaphore("/sem_example", flags=ipc.O_CREAT | ipc.O_EXCL, initial_value=1) shm_fd = os.open(shm_name, os.O_RDWR | os.O_CREAT) size = ctypes.sizeof(SharedTime) os.ftruncate(shm_fd, size) with mmap.mmap(shm_fd, length=size) as mmapped_file: st = SharedTime.from_buffer(mmapped_file) start_ts = time.time() while True: now = time.time() - start_ts semaphore.acquire() st.timestamp = now semaphore.release() time.sleep(.001) # 控制刷新速率 except Exception as e: print(e) finally: if 'memory' in locals(): memory.close() def read_from_shared_memory(shm_name="/my_shm"): semaphore = ipc.Semaphore("/sem_example") with open(shm_name, "r+b") as f: buffer = mmap.mmap(f.fileno(), length=ctypes.sizeof(SharedTime)) view = np.ndarray(shape=(1,), dtype=np.float64, buffer=buffer)[0] for i in range(5): semaphore.acquire() ts = float(view.item()) * 1e9 / 1e9 # 转换为秒级浮点数表示形式 print(f"Read timestamp: {ts:.6f}s") semaphore.release() time.sleep(1) if __name__ == "__main__": writer_process = Process(target=write_to_shared_memory) reader_process = Process(target=read_from_shared_memory) writer_process.daemon = True writer_process.start() reader_process.start() writer_process.join(timeout=7) reader_process.join() ``` 这种方式提供了更低级别的控制以及更高的性能，适用于需要频繁读写的场合。不过需要注意的是这仅限于 Linux 或 macOS 平台，并且涉及到更多复杂的错误处理逻辑。