咨询：两个后台独立Python模块的高效安全数据传输方案

嘿，用文本文件做进程间通信确实是最容易想到的入门方案，但它的弊端也很明显——IO延迟、文件锁竞争风险，还有数据被意外篡改的可能。针对你这种本地两个独立Python模块的场景，给你推荐几个更高效、安全的替代方案，按易用性和适用场景排序：

1. 轻量首选：multiprocessing 跨进程队列（Manager.Queue） #

这是Python标准库自带的方案，完全不需要额外依赖，内存级通信速度比文件快N倍，而且天然保证数据安全（只有关联进程能访问）。

因为你的两个模块是独立运行的脚本（不是父子进程），所以要用multiprocessing.Manager()创建跨进程共享的队列：

生产者模块（sender.py） #

from multiprocessing import Manager
import time

if __name__ == "__main__":
    # 创建跨进程共享队列
    with Manager() as manager:
        q = manager.Queue()
        while True:
            # 每隔3秒发送字符串
            msg = f"当前时间: {time.strftime('%Y-%m-%d %H:%M:%S')}"
            q.put(msg)
            print(f"已发送: {msg}")
            time.sleep(3)

消费者模块（receiver.py） #

from multiprocessing import Manager
import time

if __name__ == "__main__":
    with Manager() as manager:
        q = manager.Queue()
        while True:
            # 阻塞等待新消息
            msg = q.get()
            print(f"收到消息: {msg}")
            time.sleep(0.1)  # 模拟处理时间

注意：两个脚本需要同时运行，Manager会在后台启动一个服务进程来托管队列，确保跨进程可见。

2. 灵活高效：Unix域套接字（Unix Domain Socket） #

比网络套接字（TCP/UDP）快得多，因为不走网络协议栈，仅本地进程可访问，安全性拉满。适合需要更灵活通信模式（比如双向通信）的场景：

服务器模块（receiver.py） #

import socket
import os

# 定义套接字文件路径
SOCKET_PATH = "/tmp/my_ipc_socket"

# 先清理旧的套接字文件
if os.path.exists(SOCKET_PATH):
    os.unlink(SOCKET_PATH)

# 创建Unix域套接字
server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server_socket.bind(SOCKET_PATH)
server_socket.listen(1)

print("等待连接...")
conn, addr = server_socket.accept()
print("已连接")

try:
    while True:
        data = conn.recv(1024)
        if not data:
            break
        print(f"收到: {data.decode('utf-8')}")
finally:
    conn.close()
    server_socket.close()
    os.unlink(SOCKET_PATH)

客户端模块（sender.py） #

import socket
import time

SOCKET_PATH = "/tmp/my_ipc_socket"

client_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
client_socket.connect(SOCKET_PATH)

try:
    while True:
        msg = f"定时消息: {time.strftime('%Y-%m-%d %H:%M:%S')}"
        client_socket.sendall(msg.encode('utf-8'))
        print(f"已发送: {msg}")
        time.sleep(3)
finally:
    client_socket.close()

3. 可扩展方案：Redis 消息队列 #

如果以后你的需求可能扩展到多机器，或者需要消息持久化，Redis的List或Pub/Sub模式是绝佳选择。速度快，支持多生产者多消费者，还自带数据持久化：

首先需要安装Redis客户端：pip install redis

生产者模块 #

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)

while True:
    msg = f"Redis消息: {time.strftime('%Y-%m-%d %H:%M:%S')}"
    # 把消息推入队列
    r.lpush("my_ipc_queue", msg)
    print(f"已发送: {msg}")
    time.sleep(3)

消费者模块 #

import redis

r = redis.Redis(host='localhost', port=6379, db=0)

while True:
    # 阻塞等待队列消息（超时时间0表示一直等）
    _, msg = r.brpop("my_ipc_queue", timeout=0)
    print(f"收到: {msg.decode('utf-8')}")

4. 极致性能：共享内存（需手动加锁） #

如果对性能要求极高，且数据量很小（比如短字符串），可以用multiprocessing的共享内存，但需要自己处理锁机制避免竞争：

# 生产者模块
from multiprocessing import Manager
import time

if __name__ == "__main__":
    with Manager() as manager:
        # 创建共享字符串和锁
        shared_msg = manager.Value('c', b'')
        lock = manager.Lock()
        while True:
            msg = f"共享内存消息: {time.strftime('%Y-%m-%d %H:%M:%S')}".encode('utf-8')
            with lock:
                shared_msg.value = msg
            time.sleep(3)

# 消费者模块
from multiprocessing import Manager
import time

if __name__ == "__main__":
    with Manager() as manager:
        shared_msg = manager.Value('c', b'')
        lock = manager.Lock()
        last_msg = b''
        while True:
            with lock:
                current_msg = shared_msg.value
            if current_msg != last_msg:
                print(f"收到: {current_msg.decode('utf-8')}")
                last_msg = current_msg
            time.sleep(0.5)

方案对比 #

优先推荐Manager.Queue，因为它不需要额外依赖，代码最简单，完全满足你的定时传字符串需求，比文件监控高效太多。

内容的提问来源于stack exchange，提问作者Saransh Gupta