1. 需求与设计

我所设计的计算服务旨在满足多个客户对复杂计算任务的需求。由于这些计算任务通常耗时较长且资源消耗较大，为了优化客户体验并减少等待时间，我采取了并行计算的策略来显著提升计算效率。

为实现这一目标，我计划利用Python的multiprocessing库来构建一个高效的多进程计算服务。同时，为了与外部的RabbitMQ消息队列系统无缝集成，我选择了pika库作为与RabbitMQ通信的桥梁。

在计算服务的架构中，我设计了生产者和消费者两个核心角色。生产者负责监听RabbitMQ队列中的消息，每当接收到新的消息时，它将从消息中提取出待处理的数据，并将其放入multiprocessing.Queue中等待处理。这个过程充分利用了multiprocessing.Queue提供的进程间通信机制，确保数据的可靠传递和同步。

与此同时，消费者进程则负责从multiprocessing.Queue中取出待处理的数据，进行实际的计算工作。这些计算任务可能需要大约一个小时的时间来完成，取决于数据的复杂性和计算资源的分配情况。一旦计算完成，消费者进程将把结果重新封装成消息，并通过pika库发送回RabbitMQ的指定队列中，以便客户或其他系统能够获取到这些结果。

通过这种设计，我的计算服务能够同时处理多个客户的计算请求，并通过并行计算的方式提高整体的处理效率。同时，RabbitMQ的引入使得服务的可扩展性和可靠性得到了极大的提升，即使在面对高并发或系统故障的情况下，也能保证数据的完整性和服务的稳定性。

2. multiprocessing多进行模块

multiprocessing模块是Python标准库的一部分，提供了一种基于多进程的并行计算方法。通过使用multiprocessing模块，可以利用多个CPU核心来并行处理任务，从而提高程序的执行效率。与多线程相比，多进程避免了全局解释器锁（GIL）的限制，因此在CPU密集型任务中表现更优。

2.1. multiprocessing模块的核心概念和功能

• 进程（Process）：通过multiprocessing.Process类可以创建和管理独立的进程。每个进程都有自己的内存空间，互不干扰。

• 进程间通信（IPC）：multiprocessing提供了多种进程间通信的机制，如Queue、Pipe、Value和Array。在代码中，我们使用multiprocessing.Queue来在主进程和工作进程之间传递数据。

• 同步（Synchronization）：multiprocessing模块还提供了同步原语，如Lock、Semaphore、Event、Condition和Barrier，用于协调进程之间的操作。

2.2. 代码中的multiprocessing使用说明

以下是代码中使用multiprocessing的主要部分的解释：

创建和管理进程

# 创建工作进程
processes = []
for _ in range(3):p = multiprocessing.Process(target=worker, args=(data_queue, rabbitmq_params, target_queue))p.start()processes.append(p)logger.info(f'Process started with PID: {p.pid}')

• multiprocessing.Process类用于创建一个新的进程。
• target参数指定了该进程将运行的目标函数，这里是worker函数。
• args参数提供了传递给目标函数的参数。
• p.start()方法启动了进程。

进程间通信

# 创建一个multiprocessing.Queue用于进程间通信
data_queue = multiprocessing.Queue()

• multiprocessing.Queue用于在主进程和工作进程之间传递数据。它是线程和进程安全的队列，提供了先进先出（FIFO）的数据传递方式。

工作进程函数

def worker(data_queue, rabbitmq_params, target_queue):while True:try:data = data_queue.get()if data is None:breakresult = compute_result(data)try:logger.info('Worker started')connection = pika.BlockingConnection(rabbitmq_params)channel = connection.channel()except pika.exceptions.AMQPError as e:logger.error(f"Error connecting to RabbitMQ in worker: {e}")return            send_result_to_rabbitmq(channel, target_queue, result)except Exception as e:logger.error(f"An error occurred in worker: {e}")try:channel.close()connection.close()except pika.exceptions.AMQPError as e:logger.error(f"Error closing RabbitMQ connection in worker: {e}")logger.info('Worker finished')

• worker函数从data_queue中获取数据进行处理。如果队列为空，data_queue.get()将阻塞直到有数据可用。
• 处理完成后，结果通过send_result_to_rabbitmq函数发送到RabbitMQ。

终止工作进程

# 等待所有工作进程结束
for p in processes:data_queue.put(None)  # 发送None信号以终止工作进程
for p in processes:p.join()

• 向data_queue发送None信号，通知工作进程终止。
• 使用p.join()方法等待每个进程完成。这是一个阻塞调用，直到对应的进程终止。

2.3. 小结

• multiprocessing模块通过创建独立的进程，利用多核CPU的能力并行处理任务。
• multiprocessing.Queue用于进程间通信，确保数据安全地在进程间传递。
• 通过错误处理和日志记录，提高了程序的健壮性和可维护性。

这使得程序能够在多核环境中高效运行，并能够处理各种异常情况，确保程序的稳定性和可靠性。

3. 实践代码

import pika  
import multiprocessing  
import time 
from loguru import logger# 假设这是你的计算函数  
def compute_result(data):  # 模拟计算过程  time.sleep(3 * 60)  # 假设需要3分钟，换成随机  return f"Result for {data}"  # 发送结果到RabbitMQ的函数  
def send_result_to_rabbitmq(channel, queue_name, result):  try:channel.queue_declare(queue=queue_name, durable=True)channel.basic_publish(exchange='',routing_key=queue_name,body=result,properties=pika.BasicProperties(delivery_mode=2,  # make message persistent))logger.info(f"发送结果消息：{result} 到RabbitMQ")except pika.exceptions.AMQPError as e:logger.error(f"Error sending result to RabbitMQ: {e}")raise # 从RabbitMQ接收数据并放入队列的函数（生产者）  
def consume_from_rabbitmq_and_enqueue(rabbitmq_connection, rabbitmq_queue, data_queue):  try:channel = rabbitmq_connection.channel()channel.queue_declare(queue=rabbitmq_queue, durable=True)def callback(ch, method, properties, body):try:data_queue.put(body.decode('utf-8'))logger.info(f"接收到消息：{body.decode('utf-8')}")except Exception as e:logger.error(f"Error putting message into data_queue: {e}")channel.basic_consume(queue=rabbitmq_queue, on_message_callback=callback, auto_ack=True)channel.start_consuming()except pika.exceptions.AMQPError as e:logger.error(f"Error consuming from RabbitMQ: {e}")raise# 工作进程函数  
def worker(data_queue,rabbitmq_params, target_queue):  # 由于计算时间长，连接很容易被断开#try:#    logger.info('Worker started')#    workerconnection = pika.BlockingConnection(rabbitmq_params)#    channel = workerconnection.channel()#    channel.queue_declare(queue = target_queue, durable=True)#except pika.exceptions.AMQPError as e:#    logger.error(f"Error connecting to RabbitMQ in worker: {e}")#    returnwhile True:  try:  # 从Queue中获取数据，如果队列为空，则阻塞等待  data = data_queue.get() print(f'data_queue.get() is {data}') if data is None:  # 收到None信号，表示应该退出  break  # 计算结果  result = compute_result(data)  try:logger.info('Worker started')workerconnection = pika.BlockingConnection(rabbitmq_params)channel = workerconnection.channel()channel.queue_declare(queue = target_queue, durable=True)except pika.exceptions.AMQPError as e:logger.error(f"Error connecting to RabbitMQ in worker: {e}")return                          # 发送结果到RabbitMQ  send_result_to_rabbitmq(channel, target_queue, result)  except Exception as e:  print(f"An error occurred: {e}")  try:channel.close()connection.close()except pika.exceptions.AMQPError as e:logger.error(f"Error closing RabbitMQ connection in worker: {e}")logger.info('Worker finished')# 主程序  
if __name__ == "__main__":  rabbitmq_queue = 'hello_world'target_queue = 'target_station_response_queue'# 设置RabbitMQ连接和队列  credentials = pika.PlainCredentials('rabbit', '****')  # mq用户名和密码rabbitmq_params = pika.ConnectionParameters('192.168.*.*',port = 5671,virtual_host = '/dev',credentials = credentials)# 虚拟队列需要指定参数 virtual_host，如果是默认的可以不填。try:connection = pika.BlockingConnection(rabbitmq_params)except pika.exceptions.AMQPError as e:logger.error(f"Error connecting to RabbitMQ in main process: {e}")exit(1)print(' [*] Waiting for messages. To exit press CTRL+C')      # 创建一个multiprocessing.Queue用于进程间通信  data_queue = multiprocessing.Queue()  # 创建工作进程  processes = []  for _ in range(3):  p = multiprocessing.Process(target=worker, args=(data_queue, rabbitmq_params, target_queue))  p.start()  processes.append(p) print(f'process id = {p.pid}') # 假设这里是生产数据到Queue的代码（这里用模拟数据代替） channel = connection.channel()for i in range(10):  # 发送10个模拟数据  send_result_to_rabbitmq(channel, rabbitmq_queue, str(i))time.sleep(2)  # 模拟生产数据的间隔  '''# 当所有数据都生产完毕后，发送None信号给所有工作进程以结束它们  for _ in range(3):  data_queue.put(None)  '''try:consume_from_rabbitmq_and_enqueue(connection, rabbitmq_queue, data_queue)except Exception as e:logger.error(f"Error in consuming from RabbitMQ and enqueuing: {e}")      # 等待所有工作进程结束for p in processes:data_queue.put(None)  # 发送None信号以终止工作进程for p in processes:p.join()try:connection.close()except pika.exceptions.AMQPError as e:logger.error(f"Error closing RabbitMQ connection in main process: {e}")logger.info("All processes have finished.")

4. 编者按

在当今日益复杂的数据处理场景中，高效、可靠的计算服务对于企业和个人用户而言都至关重要。特别是在需要处理大量计算任务，且这些任务耗时较长、资源消耗较大的情况下，如何优化计算流程、减少用户等待时间，成为了我们必须面对的挑战。

本博客详细探讨了如何利用Python的multiprocessing库和pika库构建一个高效、可扩展的多进程计算服务，并通过RabbitMQ实现与外部系统的消息通讯。这一实践不仅解决了计算资源瓶颈问题，还显著提升了服务的整体性能和可靠性。

在整个实践过程中，我们特别感谢“文言一心”提供的指导和帮助。文言一心能够根据我们的需求快速编写出程序框架，大大提高了我们的工作效率。同时，其丰富的实践经验和专业知识也为我们提供了宝贵的参考和借鉴。同时，也感谢ChatGPT验证上述方案和专业指导。

通过本博客的分享，我们希望能够为读者提供一种高效、可靠的多进程计算服务实现方案，并希望能够帮助读者更好地理解和应用Python的multiprocessing库和pika库。在未来的数据处理和计算服务中，我们相信这一实践将发挥越来越重要的作用。

欢迎反馈。

参考：

[1]. ithicker. 多进程（多核运算）Multiprocessing. CSDN博客. 2021.02
[2]. 擒贼先擒王. Python 多进程：multiprocessing、aiomultiprocess(异步多进程). CSDN博客. 2024.05