multiprocessing多进程计算及与rabbitmq消息通讯实践

import pika
import time
import json
from multiprocessing import Process, Queue
 
# 定义一个多进程任务
def long_running_process(queue):
    # 假设这是一个耗时的计算任务
    result = do_some_long_running_computation()
    queue.put(result)  # 将结果放入进程间通信的队列中
 
# 定义一个计算任务，模拟耗时计算
def do_some_long_running_computation():
    return "任务处理结果"
 
# 定义一个回调函数，用于处理RabbitMQ发送的消息
def callback(ch, method, properties, body):
    # 将接收到的消息转换为字典
    message = json.loads(body)
    # 创建一个进程并传入消息数据
    p = Process(target=long_running_process, args=(Queue(),))
    p.start()
    
    # 处理其他业务逻辑...
    # 假设这里是将计算结果发送回RabbitMQ
    p.join()  # 等待进程完成
    response = p.get()  # 从队列中获取结果
    ch.basic_publish(exchange='',
                     routing_key=method.reply_to,  # 应答队列名称
                     properties=pika.BasicProperties(correlation_id = \
                                                     method.correlation_id),
                     body=json.dumps(response))  # 发送处理结果
    ch.basic_ack(delivery_tag=method.delivery_tag)  # 发送确认消息
 
# 连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
 
# 定义一个队列用于接收RPC响应
result = Queue()
 
# 定义一个RabbitMQ RPC服务器
channel.basic_consume(callback, queue='rpc_queue')
 
print(" [x] Awaiting RPC requests")
channel.start_consuming()

这个代码实例展示了如何使用multiprocessing库来创建多进程任务，以及如何使用RabbitMQ进行进程间通信和异步任务处理。在long_running_process函数中，我们模拟了一个耗时的计算任务，并将结果通过进程间队列传递给了回调函数。在回调函数中，我们创建了一个新的进程来处理任务，并将结果发送回客户端。这种模式可以有效提高系统的处理能力和响应速度。