第14章-Reactor框架QPS性能测试

接下来我们写一个测试用例来测一下我们的Reactor框架的qps。

qps: (Query Per Second)每秒查询率QPS是对一个特定的查询服务器在规定时间内所处理流量多少的衡量标准。

14.1 测试用例代码编写

    我们首先定义一个proto文件,用来承载客户端和服务端通信媒介的。

example/qps_test/echoMessage.proto

syntax = "proto3";
package qps_test;
message EchoMessage 
{
    string content = 1;
    int32 id = 2;
};

    然后生成对应的cc文件和h文件

protoc --cpp_out=. ./*.proto

    接下来我们来实现服务端，服务端主要就是简单的回显，客户端发什么数据，回显就可以了。

example/qps_test/server.cpp

#include <string>
#include <string.h>
#include "config_file.h"
#include "tcp_server.h"
#include "echoMessage.pb.h"
//回显业务的回调函数
void callback_busi(const char *data, uint32_t len, int msgid, net_connection *conn, void *user_data)
{
    qps_test::EchoMessage request, response;  
    //解包，确保data[0-len]是一个完整包
    request.ParseFromArray(data, len); 
    //设置新pb包
    response.set_id(request.id());
    response.set_content(request.content());
    //序列化
    std::string responseString;
    response.SerializeToString(&responseString);
    conn->send_message(responseString.c_str(), responseString.size(), msgid);
}
int main() 
{
    event_loop loop;
    //加载配置文件
    config_file::setPath("./serv.conf");
    std::string ip = config_file::instance()->GetString("reactor", "ip", "0.0.0.0");
    short port = config_file::instance()->GetNumber("reactor", "port", 8888);
    printf("ip = %s, port = %d\n", ip.c_str(), port);
    tcp_server server(&loop, ip.c_str(), port);
    //注册消息业务路由
    server.add_msg_router(1, callback_busi);
    loop.event_process();
    return 0;
}

接下来是客户端，客户端我们创建一个Qps结构体，来记录每秒，服务端成功回显数据的次数，来做qps统计，客户端我们可以指定开多少个线程去压测服务端。

example/qps_test/client.cpp

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <string>
#include "tcp_client.h"
#include "echoMessage.pb.h"
struct Qps
{
    Qps() {
        last_time = time(NULL); 
        succ_cnt = 0;
    }
    long last_time;//最后一次发包时间 ms为单位
    int succ_cnt; //成功收到服务器回显的次数
};
//客户端业务
void busi(const char *data, uint32_t len, int msgid, net_connection  *conn, void *user_data)
{
    Qps *qps = (Qps*)user_data; //用户参数
    qps_test::EchoMessage request, response;
    //解析服务端传来的pb数据
    if (response.ParseFromArray(data, len) == false) {
        printf("server call back data error\n");
        return;
    }
    //判断数据内容是否回显一致
    if (response.content() == "Hello Lars!!!") {
        //服务器请求响应成功一次
        qps->succ_cnt ++;
    }
    long current_time = time(NULL);
    if (current_time - qps->last_time >= 1) {
        //如果当前时间比最后记录时间大于1秒，那么我们进行记录
        printf("---> qps = %d <----\n", qps->succ_cnt);
        qps->last_time = current_time;//记录最后时间
        qps->succ_cnt = 0;//清空成功次数
    }
    //给服务端发送新的请求
    request.set_id(response.id() + 1);
    request.set_content(response.content());
    std::string requestString;
    request.SerializeToString(&requestString);
    conn->send_message(requestString.c_str(), requestString.size(), msgid);
}
//创建链接成功之后
void connection_start(net_connection *client, void *args)
{
    qps_test::EchoMessage request;
    request.set_id(1);
    request.set_content("Hello Lars!!!");
    std::string requestString;
    request.SerializeToString(&requestString);
    int msgid = 1;//与server端的消息路由一致
    client->send_message(requestString.c_str(), requestString.size(), msgid);
}
void *thread_main(void *args)
{
    //给服务端发包
    event_loop loop; 
    tcp_client client(&loop, "127.0.0.1", 7777, "qps client");
    Qps qps;
    //设置回调
    client.add_msg_router(1, busi, (void*)&qps);
    //设置链接创建成功之后Hook
    client.set_conn_start(connection_start);
    loop.event_process();
    return NULL;
}
int main(int argc, char **argv) 
{
    if (argc == 1) {
        printf("Usage: ./client [threadNum]\n");
        return 1;
    }
    //创建N个线程
    int thread_num = atoi(argv[1]);
    pthread_t *tids;
    tids = new pthread_t[thread_num];
    for (int i = 0; i < thread_num; i++) {
        pthread_create(&tids[i], NULL, thread_main, NULL);
    }
    for (int i = 0; i < thread_num; i++) {
        pthread_join(tids[i], NULL);
    }
    return 0;
}

接下来我们的Makefile

CXX=g++
CFLAGS=-g -O2 -Wall -fPIC -Wno-deprecated 
INC=-I../../include
LIB=-L../../lib -llreactor -lpthread -lprotobuf
OBJS = $(addsuffix .o, $(basename $(wildcard *.cc)))
all:
        $(CXX) -o server $(CFLAGS)  server.cpp echoMessage.pb.cc $(INC) $(LIB)
        $(CXX) -o client $(CFLAGS)  client.cpp echoMessage.pb.cc $(INC) $(LIB)
clean:
        -rm -f *.o server client

    记住编译加上`-lprotobuf` 编译的文件加上`echoMessage.pb.cc`文件。

14.2 并发测试结果

    启动服务端，再启动客户端，(注意问了方便看结果，将之前reactor的一些stdout的调试日志都关闭)

看客户端运行结果

$ ./client 1
msg_router init...
do_connect EINPROGRESS
add msg cb msgid = 1
connect 127.0.0.1:7777 succ!
---> qps = 6875 <----
---> qps = 8890 <----
---> qps = 8354 <----
---> qps = 9078 <----
---> qps = 8933 <----
---> qps = 9043 <----
---> qps = 8743 <----
---> qps = 9048 <----
---> qps = 8987 <----
---> qps = 8742 <----
---> qps = 8720 <----
---> qps = 8795 <----
---> qps = 8889 <----
---> qps = 9034 <----
---> qps = 8669 <----
---> qps = 9001 <----
---> qps = 8810 <----
---> qps = 8850 <----
---> qps = 8802 <----
---> qps = 9072 <----
---> qps = 8853 <----
---> qps = 8804 <----
---> qps = 8574 <----
---> qps = 8645 <----
---> qps = 8085 <----
---> qps = 8720 <----
...

 这里我们客户端是开启一个线程进行测试，平均每秒服务端会响应8700次左右。
    这里我简单用两个主机，分别测试了一些数据

主机1：

CPU个数：2个 ， 内存: 2GB , 系统：Ubuntu18.04虚拟机

主机2：

CPU个数: 24个 , 内存:128GB， 系统: 云主机

    现在我们的Lars-Reactor模块基本已经开完完成了，后续可能会再添加一些模块，为了其他模块方便使用Lars-Reactor的一些接口，我们可以对外提供一个公共的头文件，方便一次性导入

lars_reactor/include/lars_reactor.h

#pragma once
#include "io_buf.h"
#include "buf_pool.h"
#include "reactor_buf.h"
#include "net_connection.h"
#include "tcp_conn.h"
#include "tcp_server.h"
#include "tcp_client.h"
#include "udp_server.h"
#include "udp_client.h"
#include "message.h"
#include "task_msg.h"
#include "event_loop.h"
#include "thread_pool.h"
#include "thread_queue.h"
#include "config_file.h"