import time
import random
from multiprocessing import Process
# 模拟发送消息的函数
def send_message(queue, num_msgs):
for i in range(num_msgs):
# 模拟消息体
message = f"message_{i}"
queue.put(message)
# 模拟发送延迟
time.sleep(random.uniform(0, 0.1))
# 模拟接收消息的函数
def receive_message(queue):
while True:
message = queue.get()
# 模拟处理延迟
time.sleep(random.uniform(0.01, 0.1))
# 处理完毕后，通知队列
queue.task_done()
# 性能测试函数
def performance_test(queue, num_messages, num_workers):
start_time = time.time()
# 创建工作进程
workers = [Process(target=receive_message, args=(queue,)) for _ in range(num_workers)]
# 启动工作进程
for worker in workers:
worker.start()
# 发送消息
send_message(queue, num_messages)
# 等待所有任务完成
queue.join()
end_time = time.time()
# 计算总时间
total_time = end_time - start_time
# 输出结果
print(f"Total time taken: {total_time} seconds")
# 使用示例
if __name__ == "__main__":
from multiprocessing import Queue
queue = Queue()
num_messages = 10000 # 假设我们发送10000条消息
num_workers = 5 # 使用5个工作进程
performance_test(queue, num_messages, num_workers)这段代码模拟了一个简单的异步消息队列处理流程，其中包含发送消息、接收消息和性能测试的函数。通过多进程队列，我们可以在生产者和消费者之间建立一个高效的消息传递机制，并通过性能测试来评估系统的整体性能。