Python进程间通信与消息队列

Python进程间通信与消息队列

一、进程间通信概述

进程间通信（IPC）是多进程协作的基础。Python提供了多种IPC机制：
- 管道（Pipe）
- 队列（Queue）
- 共享内存（SharedMemory）
- 信号（Signal）
- 套接字（Socket）
- 内存映射文件（mmap）


二、multiprocessing.Queue

from multiprocessing import Process, Queue
import time

class TaskQueue:
"""基于多进程队列的任务分发系统"""
def __init__(self, num_workers=4):
self.task_queue = Queue()
self.result_queue = Queue()
self.num_workers = num_workers
self.workers = []

def start(self):
for i in range(self.num_workers):
p = Process(target=self._worker, args=(i,))
p.daemon = True
p.start()
self.workers.append(p)

def _worker(self, worker_id):
while True:
task = self.task_queue.get()
if task is None: # 毒丸信号
break
task_id, func, args, kwargs = task
try:
result = func(*args, **kwargs)
self.result_queue.put((task_id, 'success', result))
except Exception as e:
self.result_queue.put((task_id, 'error', str(e)))

def submit(self, task_id, func, *args, **kwargs):
self.task_queue.put((task_id, func, args, kwargs))

def get_result(self, timeout=None):
return self.result_queue.get(timeout=timeout)

def shutdown(self):
for _ in self.workers:
self.task_queue.put(None)
for p in self.workers:
p.join()

# 使用
def heavy_computation(n):
return sum(i * i for i in range(n))

tq = TaskQueue(num_workers=4)
tq.start()

for i in range(10):
tq.submit(i, heavy_computation, 1000000)

for i in range(10):
task_id, status, result = tq.get_result()
print(f"任务 {task_id}: {status} = {result}")

tq.shutdown()


三、Pipe管道

from multiprocessing import Process, Pipe

def sender(conn, messages):
for msg in messages:
conn.send(msg)
time.sleep(0.1)
conn.send(None) # 结束信号
conn.close()

def receiver(conn):
while True:
msg = conn.recv()
if msg is None:
break
print(f"收到: {msg}")
conn.close()

# Pipe返回两个连接对象
parent_conn, child_conn = Pipe()

p1 = Process(target=sender, args=(parent_conn, ["hello", "world", "python"]))
p2 = Process(target=receiver, args=(child_conn,))

p1.start()
p2.start()
p1.join()
p2.join()


四、共享内存（Python 3.8+）

from multiprocessing import shared_memory, Process
import numpy as np

def worker_with_shared_memory(shm_name, shape, dtype):
"""工作进程访问共享内存"""
existing_shm = shared_memory.SharedMemory(name=shm_name)
array = np.ndarray(shape, dtype=dtype, buffer=existing_shm.buf)

# 直接修改共享数组
array *= 2

existing_shm.close()

# 主进程创建共享内存
data = np.array([1, 2, 3, 4, 5], dtype=np.int64)
shm = shared_memory.SharedMemory(create=True, size=data.nbytes)
shared_array = np.ndarray(data.shape, dtype=data.dtype, buffer=shm.buf)
shared_array[:] = data[:]

# 启动工作进程
p = Process(target=worker_with_shared_memory,
args=(shm.name, data.shape, data.dtype))
p.start()
p.join()

print(shared_array) # [2, 4, 6, 8, 10]

# 清理
shm.close()
shm.unlink()


五、Manager对象

from multiprocessing import Manager, Process

def worker(shared_dict, shared_list, lock, worker_id):
with lock:
shared_dict[f'worker_{worker_id}'] = f'result_{worker_id}'
shared_list.append(worker_id)

# Manager提供进程安全的共享对象
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, args=(shared_dict, shared_list, lock, i))
processes.append(p)
p.start()

for p in processes:
p.join()

print(dict(shared_dict))
print(list(shared_list))


六、信号处理

import signal
import sys

class GracefulShutdown:
"""优雅关闭处理"""
def __init__(self):
self.shutdown_requested = False
signal.signal(signal.SIGTERM, self._handle_signal)
signal.signal(signal.SIGINT, self._handle_signal)

def _handle_signal(self, signum, frame):
print(f"\n收到信号 {signum}，准备关闭...")
self.shutdown_requested = True

def should_continue(self):
return not self.shutdown_requested

# 使用
shutdown = GracefulShutdown()

while shutdown.should_continue():
# 执行工作
process_next_task()
time.sleep(1)

print("清理资源...")
cleanup()
print("已安全关闭")


七、简易消息队列实现

import json
import threading
from queue import Queue, Empty
from typing import Callable, Any

class MessageBroker:
"""简易发布/订阅消息代理"""
def __init__(self):
self._topics = {}
self._subscribers = {}
self._lock = threading.Lock()

def create_topic(self, topic_name, max_size=1000):
with self._lock:
if topic_name not in self._topics:
self._topics[topic_name] = Queue(maxsize=max_size)
self._subscribers[topic_name] = []

def publish(self, topic_name, message):
if topic_name not in self._topics:
raise ValueError(f"主题不存在: {topic_name}")
self._topics[topic_name].put(message)
self._notify_subscribers(topic_name, message)

def subscribe(self, topic_name, callback: Callable):
if topic_name not in self._subscribers:
raise ValueError(f"主题不存在: {topic_name}")
self._subscribers[topic_name].append(callback)

def _notify_subscribers(self, topic_name, message):
for callback in self._subscribers[topic_name]:
threading.Thread(target=callback, args=(message,)).start()

def consume(self, topic_name, timeout=None):
if topic_name not in self._topics:
raise ValueError(f"主题不存在: {topic_name}")
try:
return self._topics[topic_name].get(timeout=timeout)
except Empty:
return None

# 使用
broker = MessageBroker()
broker.create_topic('orders')
broker.create_topic('notifications')

# 订阅
broker.subscribe('orders', lambda msg: print(f"处理订单: {msg}"))

# 发布
broker.publish('orders', {'order_id': 1, 'amount': 99.99})


八、使用Redis作为消息队列

class RedisMessageQueue:
"""基于Redis的消息队列（示意实现）"""
def __init__(self, redis_client, queue_name):
self.redis = redis_client
self.queue_name = queue_name

def publish(self, message):
data = json.dumps(message)
self.redis.lpush(self.queue_name, data)

def consume(self, timeout=0):
result = self.redis.brpop(self.queue_name, timeout=timeout)
if result:
_, data = result
return json.loads(data)
return None

def consume_batch(self, batch_size=10):
pipe = self.redis.pipeline()
for _ in range(batch_size):
pipe.rpop(self.queue_name)
results = pipe.execute()
return [json.loads(r) for r in results if r]

def length(self):
return self.redis.llen(self.queue_name)

class RedisPubSub:
"""Redis发布/订阅"""
def __init__(self, redis_client):
self.redis = redis_client
self.pubsub = redis_client.pubsub()

def subscribe(self, channel, callback):
self.pubsub.subscribe(**{channel: callback})
thread = self.pubsub.run_in_thread(sleep_time=0.01)
return thread

def publish(self, channel, message):
self.redis.publish(channel, json.dumps(message))


九、异步消息处理

import asyncio

class AsyncMessageQueue:
"""异步消息队列"""
def __init__(self, maxsize=0):
self.queue = asyncio.Queue(maxsize=maxsize)
self.handlers = []
self._running = False

def register_handler(self, handler):
self.handlers.append(handler)

async def publish(self, message):
await self.queue.put(message)

async def start_consuming(self, num_consumers=3):
self._running = True
consumers = [
asyncio.create_task(self._consumer(f"consumer-{i}"))
for i in range(num_consumers)
]
await asyncio.gather(*consumers)

async def _consumer(self, name):
while self._running:
try:
message = await asyncio.wait_for(self.queue.get(), timeout=1.0)
for handler in self.handlers:
try:
await handler(message)
except Exception as e:
print(f"{name} 处理消息失败: {e}")
self.queue.task_done()
except asyncio.TimeoutError:
continue

async def stop(self):
self._running = False
await self.queue.join()

# 使用
async def order_handler(message):
print(f"处理订单: {message}")
await asyncio.sleep(0.5)

async def main():
mq = AsyncMessageQueue()
mq.register_handler(order_handler)

# 启动消费者
consumer_task = asyncio.create_task(mq.start_consuming(3))

# 发布消息
for i in range(10):
await mq.publish({'order_id': i, 'amount': i * 10})

await mq.queue.join()
await mq.stop()


十、总结

IPC选择建议：
- 简单数据传递 -> Queue或Pipe
- 大量数据共享 -> SharedMemory
- 复杂共享对象 -> Manager
- 分布式系统 -> Redis/RabbitMQ/Kafka
- 异步场景 -> asyncio.Queue

设计原则：
1. 优先使用Queue，它是进程安全的
2. 共享内存性能最高但需要手动同步
3. 消息队列解耦生产者和消费者
4. 考虑消息丢失、重复消费、顺序性问题
5. 生产环境使用成熟的消息中间件