本文目录导读：
Python 实现进程间通信（IPC, Inter-Process Communication）有多种方式，我会详细介绍最常用的几种方法,并给出完整的示例代码。
使用 multiprocessing 模块（最常用）
1 Queue（队列）
from multiprocessing import Process, Queue
import time
def producer(queue):
"""生产者进程"""
for i in range(5):
msg = f"消息 {i}"
queue.put(msg)
print(f"生产者发送: {msg}")
time.sleep(1)
queue.put("STOP") # 发送结束信号
def consumer(queue):
"""消费者进程"""
while True:
msg = queue.get()
if msg == "STOP":
break
print(f"消费者收到: {msg}")
time.sleep(0.5)
if __name__ == "__main__":
# 创建队列
queue = Queue()
# 创建并启动进程
p1 = Process(target=producer, args=(queue,))
p2 = Process(target=consumer, args=(queue,))
p1.start()
p2.start()
p1.join()
p2.join()
2 Pipe（管道）
from multiprocessing import Process, Pipe
import time
def sender(conn):
"""发送端"""
messages = ["Hello", "World", "Python", "IPC", "END"]
for msg in messages:
conn.send(msg)
print(f"发送: {msg}")
time.sleep(0.5)
conn.close()
def receiver(conn):
"""接收端"""
while True:
try:
msg = conn.recv()
if msg == "END":
break
print(f"接收: {msg}")
except EOFError:
break
conn.close()
if __name__ == "__main__":
# 创建管道 (duplex=True 表示双向通信)
parent_conn, child_conn = Pipe(duplex=True)
p1 = Process(target=sender, args=(parent_conn,))
p2 = Process(target=receiver, args=(child_conn,))
p1.start()
p2.start()
p1.join()
p2.join()
3 Shared Memory（共享内存）
from multiprocessing import Process, Value, Array
import time
def increment(shared_value, shared_array):
"""修改共享数据"""
for _ in range(10):
# 修改共享值
shared_value.value += 1
# 修改共享数组
shared_array[0] += 1
shared_array[1] += 2
print(f"Value: {shared_value.value}, Array: {shared_array[:]}")
time.sleep(0.5)
def read_values(shared_value, shared_array):
"""读取共享数据"""
for _ in range(5):
print(f"读取 - Value: {shared_value.value}, Array: {shared_array[:]}")
time.sleep(1)
if __name__ == "__main__":
# 创建共享内存
shared_value = Value('i', 0) # 'i' 表示整数类型
shared_array = Array('i', [0, 0, 0]) # 整数数组
p1 = Process(target=increment, args=(shared_value, shared_array))
p2 = Process(target=read_values, args=(shared_value, shared_array))
p1.start()
p2.start()
p1.join()
p2.join()
使用 socket（网络通信）
import socket
import multiprocessing
import time
def server_process():
"""服务器进程"""
# 创建UNIX域套接字
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server.bind('/tmp/test_socket')
server.listen(1)
conn, addr = server.accept()
while True:
data = conn.recv(1024)
if not data:
break
print(f"服务器收到: {data.decode()}")
conn.send(f"ACK: {data.decode()}".encode())
conn.close()
server.close()
def client_process():
"""客户端进程"""
time.sleep(0.1) # 等待服务器启动
client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
client.connect('/tmp/test_socket')
messages = ["Hello", "World", "END"]
for msg in messages:
client.send(msg.encode())
response = client.recv(1024)
print(f"客户端收到: {response.decode()}")
time.sleep(0.5)
client.close()
if __name__ == "__main__":
import os
try:
os.unlink('/tmp/test_socket')
except OSError:
pass
p1 = multiprocessing.Process(target=server_process)
p2 = multiprocessing.Process(target=client_process)
p1.start()
p2.start()
p1.join()
p2.join()
使用文件（File-based IPC）
import time
import multiprocessing
import os
def writer_process(filename):
"""写入进程"""
for i in range(5):
with open(filename, 'w') as f:
f.write(f"Message {i}\n")
print(f"写入: Message {i}")
time.sleep(1)
def reader_process(filename):
"""读取进程"""
for _ in range(10):
if os.path.exists(filename):
with open(filename, 'r') as f:
content = f.read().strip()
if content:
print(f"读取: {content}")
time.sleep(0.5)
if __name__ == "__main__":
filename = "ipc_temp.txt"
p1 = multiprocessing.Process(target=writer_process, args=(filename,))
p2 = multiprocessing.Process(target=reader_process, args=(filename,))
p1.start()
p2.start()
p1.join()
p2.join()
# 清理临时文件
if os.path.exists(filename):
os.remove(filename)
使用 Manager（高级同步）
from multiprocessing import Process, Manager
import time
def worker_process(shared_dict, shared_list, lock):
"""工作进程"""
with lock:
shared_dict['count'] = shared_dict.get('count', 0) + 1
shared_list.append(time.time())
print(f"Worker added: {shared_dict['count']}")
if __name__ == "__main__":
with Manager() as manager:
# 创建共享数据结构
shared_dict = manager.dict()
shared_list = manager.list()
lock = manager.Lock()
processes = []
for i in range(5):
p = Process(target=worker_process, args=(shared_dict, shared_list, lock))
processes.append(p)
p.start()
for p in processes:
p.join()
print(f"Final dict: {shared_dict}")
print(f"Final list: {shared_list}")
选择建议
| 方法 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| Queue | 生产者-消费者模式 | 线程安全，易用 | 需要序列化 |
| Pipe | 点对点通信 | 速度快，双向 | 仅两个进程 |
| Shared Memory | 大量数据共享 | 高效，无需序列化 | 复杂数据类型限制 |
| Socket | 跨机器通信 | 通用性强 | 需要序列化 |
| Manager | 复杂数据结构 | 支持多种数据类型 | 性能相对较慢 |
完整示例：实时数据采集系统
from multiprocessing import Process, Queue, Value, Manager
import random
import time
import signal
class DataCollector:
"""数据采集系统示例"""
def __init__(self):
self.data_queue = Queue()
self.running = Value('b', True) # 布尔类型共享变量
self.stats = Manager().dict({
'total_collected': 0,
'error_count': 0
})
def collector(self, sensor_id):
"""数据采集进程"""
while self.running.value:
try:
# 模拟采集数据
data = {
'sensor_id': sensor_id,
'value': random.gauss(100, 15),
'timestamp': time.time()
}
self.data_queue.put(data)
self.stats['total_collected'] += 1
print(f"采集器 {sensor_id}: {data['value']:.2f}")
time.sleep(random.uniform(0.5, 1.5))
except Exception as e:
self.stats['error_count'] += 1
print(f"采集器 {sensor_id} 错误: {e}")
def processor(self):
"""数据处理进程"""
processed_count = 0
while self.running.value or not self.data_queue.empty():
try:
data = self.data_queue.get(timeout=1)
# 处理数据
data['processed_value'] = data['value'] * 1.1
data['processed'] = True
processed_count += 1
print(f"处理: 传感器 {data['sensor_id']} -> {data['processed_value']:.2f}")
except:
pass
print(f"总处理数据: {processed_count}")
def monitor(self):
"""监控进程"""
while self.running.value:
print(f"状态: 采集 {self.stats['total_collected']}, 错误 {self.stats['error_count']}")
time.sleep(3)
def start(self, duration=20):
"""启动系统"""
# 创建3个采集进程
collectors = [
Process(target=self.collector, args=(i,))
for i in range(3)
]
# 创建处理进程
processor = Process(target=self.processor)
# 创建监控进程
monitor = Process(target=self.monitor)
# 启动所有进程
for p in collectors + [processor, monitor]:
p.start()
# 运行指定时间
time.sleep(duration)
# 停止系统
self.running.value = False
# 等待所有进程结束
for p in collectors + [processor, monitor]:
p.join()
print("系统停止")
print(f"最终统计: {dict(self.stats)}")
if __name__ == "__main__":
system = DataCollector()
system.start(duration=15)
这些示例涵盖了Python中主要的进程间通信方式,选择哪种方式取决于你的具体需求：
- 简单消息传递：使用
Queue或Pipe - 大量数据共享：使用
Shared Memory - 复杂数据结构：使用
Manager - 跨机器通信：使用
Socket - 实时性要求高：使用
Pipe或Shared Memory
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
