RabbitMq性能测试工具

公司需要对mq总线进行测试，需要对消息协议进行解析并统计tps和响应时间，网上没有好的轮子，自己写一个。设计思路：通过mq收发消息，在消息体中添加send_time和msg_seq字段用于异步场景下统计消息的响应时间，对测试的结果进行可视化。测试结果：空载的时候，发送和接收各10K次/s。
代码如下：

# RabbitMq性能压测脚本,依赖pika,np,pandas,pyecharts模块--2019.11.26
import os
import time
from multiprocessing import Queue, Process, freeze_support,Event
import pika
import json
import traceback
import numpy as np
from queue import Empty
import pandas as pd
from pyecharts import options as opts
from pyecharts.charts import Line, Page


class data_polat():
    """
    将数据可式化
    """

    def __init__(self, file_name="data"):
        """
        :param file_name:存储文件的名称不带尾缀
        """
        self.dir = os.path.join(__file__, '..', 'data')
        if not os.path.isdir(self.dir):
            os.mkdir(self.dir)
        self.file_path = os.path.join(self.dir)
        self.file_name = file_name

        self.data_dict = dict()


    def input(self, data_list):
        """
        输入要可视化的数据
        :parma data_df DataFrame对象
        :param data_list:汇总的数据列变 eg: [
        {'total': 51517, 'max': 8.975, 'mean': 1.412, 'meadian': 1.412, 'std': 0.746},
        {'total': 52281, 'max': 7.988, 'mean': 1.343, 'meadian': 1.343, 'std': 0.688},
        {'total': 52264, 'max': 6.017, 'mean': 1.324, 'meadian': 1.324, 'std': 0.59},
        ]
        数据保存为csv文件
        """
        df = pd.DataFrame(data_list)
        # df.to_json(os.path.join(self.dir,"data.json"))
        path = os.path.join(self.dir, str(self.file_name) + ".csv")
        df.to_csv(path)
        print(f"测试结果的csv文件保存路径：{path}")
        title_list = list(df.columns)
        for key in title_list:
            self.data_dict[key] = list(df.loc[:, key])

    def render_line(self):
        """
        绘制测试结果的折现图
        :param data_list:
        :param data_dict:
        :return:
        """
        # 获取采样周期个数
        lenth = len(list(self.data_dict.values())[0])
        page = Page()
        columns = [i for i in range(lenth)]
        for key, value_list in self.data_dict.items():
            line = Line()
            line.add_xaxis(columns)
            line.add_yaxis(key, value_list, is_symbol_show=True)
            line.set_global_opts(opts.TitleOpts(title=key))
            page.add(line)
        path = os.path.join(self.dir, str(self.file_name) + ".html")
        page.render(path)
        print(f"html文件保存路径：{path}")


class mq_test():
    """
    服务的消费模式时：生产-消息的单点消费模式
    多进程执行测试任务：
    1.pub_process 一些进程通过pika发送mq消息；
    2.sub_process 一些进程订阅推送频道的消息，并按规则推送到统计进程;
    3.一个统计进程进程，对结果做实时的处理（利用pandas）并存储；
        运行结束后调用:测试结果的可式化，对统计进程存储的数据，进行可视化；
    """

    def __init__(self,
                 host="192.168.52.72",
                 port=5672,
                 user="admin",
                 passwd="admin",
                 data_name="data"):

        self.host = host
        self.user = user
        self.port = port
        self.passwd = passwd
        self.result_queue = Queue()  # 进程间通信的队列
        self.data_name = data_name
        self.msg_seq = 1  # 消息的推送序列号
        self.data_list = []  # 用于存储每个采样周期的的数据
        self.event = Event()

    def __build_channel(self, topic):
        """
        创建一个channel对象，用于发送或者订阅
        :return:
        """
        credentials = pika.PlainCredentials(self.user, self.passwd)
        connection = pika.BlockingConnection(pika.ConnectionParameters(self.host, self.port, '/', credentials))
        channel = connection.channel()
        channel.queue_declare(queue=topic, durable=True)
        return channel

    def reg_msg_fun(self, fun, *args, **kwargs):
        """
        注册一个生产发送数据的函数，fun的返回值必须是一个可迭代对象，元素为字典
        :param fun:
        :return:
        """
        self.__args = args
        self.__kwargs = kwargs
        self.__msg_fun = fun

    def __build_msg(self):
        """
        生成消息对象
        :return:
        """
        msg_iter = self.__msg_fun(*self.__args, **self.__kwargs)
        return msg_iter

    def pub_process(self, pub_topic):
        """
        发送mq消息的进程
        :param msg 发送的消息可迭代对象
        :return:
        """
        channel = self.__build_channel(pub_topic)
        msg_iter = self.__build_msg()
        self.event.set()
        print("初始化完成，开始发送消息")
        for msg in msg_iter:
            try:
                msg["send_time"] = time.time()
                body = json.dumps(msg)
            except Exception:
                print("消息json序列化失败")
                continue
            try:
                channel.basic_publish(exchange="",
                                      routing_key=pub_topic,
                                      body=body)
                # time.sleep(2)
            except Exception as e:
                print(f"消息发送异常")
        else:
            print("数据发送完毕，退出发送进程")

    def sub_process(self, sub_topic):
        """
        订阅指定频道的进程
        :param sub_topic:
        :return:
        """
        self.event.wait()
        print("发送进程已运行，开始订阅进程")
        def callback(ch, method, properties, body):
            self.result_queue.put(body)

        channel = self.__build_channel(sub_topic)
        channel.basic_consume(sub_topic, callback, auto_ack=True)
        channel.start_consuming()

    def statas_process(self, interval=5, time_out=60):
        """
        汇总单个采样周期类的数据，统计结束后将数据可视化
        :param interval: 采样间隔，默认为5秒，建议大于5秒；
        :param time_out: 获取测试结果超时退出进程的时间，建议大于60s
        :return:
        """
        self.event.wait()
        print("发送进程已运行，结果分析进程开始运行")
        run_time = time.time()
        start_time = time.time()
        result_list = []
        end_tag = False
        while 1:
            cost_time = None
            try:
                data_str = self.result_queue.get(timeout=time_out)
                msg = json.loads(data_str)
                cost_time = time.time() - msg.get("send_time")
                end_time = time.time()
            except Empty:
                self.run_cost_time = time.time() - run_time - time_out
                print(f"数据分析完毕")
                end_tag = True
            except Exception as e:
                print(f"当次数据解析异常,跳过执行:{traceback.format_exc()}")

            if end_tag is False and end_time - start_time < interval:
                result_list.append(cost_time)
            else:
                if cost_time:
                    result_list.append(cost_time)
                result = self.ans_result(result_list)
                self.data_list.append(result)
                print(result)
                start_time = end_time
                result_list = []

            if end_tag is True:
                print("统计进程退出")
                self.__summary()
                break

    def ans_result(self, result_list):
        """
        统计一个采样周期内的：采样数，最大值，平均值，中位值，标准差
        :param result_list:
        :return:
        """
        ans_method = {'total': len, 'max': np.max, 'mean': np.mean, 'meadian': np.nanmean, 'std': np.std}
        data_ary = np.array(result_list) * 1000  # 转换成数组，值转换为毫秒
        result_dict = dict()
        for k, v in ans_method.items():
            result_dict[k] = round(v(data_ary), 3)
        return result_dict

    def __summary(self):
        """
        汇总测试结果
        :return:
        """
        df = pd.DataFrame(self.data_list)
        total = df['total'].sum()
        total_mean = round(total / self.run_cost_time, 3)
        rsp_max = round(df['max'].max(), 3)
        rsp_mean = round(df['mean'].mean(), 3)

        print(f"测试结果概览:\n"
              f"响应的总消息数：{total}\n"
              f"最大的响应时间(ms)：{rsp_max}\n"
              f"平均响应时间(ms):{rsp_mean}\n"
              f"运行时间(s):{round(self.run_cost_time,3)}\n"
              f"每秒的平均响应次数：{total_mean}")

        line = data_polat(self.data_name)
        line.input(self.data_list)
        line.render_line()


def msg_fun():
    """
    生成消息的函数
    :return:
    """
    return ({'test': 'test'} for i in range(10 ** 8))


if __name__ == "__main__":
    freeze_support()
    mq = mq_test()
    pub_topic = "topic.test.api"
    sub_topic = "topic.test.api"
    mq.reg_msg_fun(msg_fun)
    job_list = []
    job_list.append(Process(target=mq.pub_process, args=(pub_topic,)))
    job_list.append(Process(target=mq.sub_process, args=(sub_topic,)))
    job_list.append(Process(target=mq.statas_process))
    for job in job_list:
        job.daemon = True
        job.start()
    job_list[-1].join()  # 以统计线程为退出的标志

image.png