MQ异步消息架构性能测试及瓶颈分析

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)

这段代码模拟了一个简单的异步消息队列处理流程，其中包含发送消息、接收消息和性能测试的函数。通过多进程队列，我们可以在生产者和消费者之间建立一个高效的消息传递机制，并通过性能测试来评估系统的整体性能。