1. 背景

昨天看到一段brpc中的压测代码rpc_press, 看着不错。整理一下。
发压工具的难点不是发送请求，而是要注意下面的2点:

• 保证能发出足够的qps，比如上万qps
• 控制发送合理的qps，比如控制为5qps，不可以大量发压

2. brpc 中的是关键实现

2.1 如何确保发送足够qps

rpc_press 采用多线程发送。
对于上万qps，多线程来分摊，每个线程发送qps控制在一定范围内。

2.2 如何控制合理qps

这是难点，线程如何控制发送qps。
看下面公式，1000000us(1s)需要发送200请求, 那50请求需要多少时间了？这就是brpc控制压力的核心原理。

核心原理: rpc_press中根据目前发送请求的速度与应该的速度进行比较，来判断是继续发压，还是usleep

2.3 rpc_press 代码

这里贴一点rpc_press工作线程git代码(引用自brpc)

void RpcPress::sync_client() { 
    double req_rate = _options.test_req_rate / _options.test_thread_num; 
    //max make up time is 5 s 
    if (_msgs.empty()) { 
        LOG(ERROR) << "nothing to send!"; 
        return; 
    } 
    const int thread_index = g_thread_count.fetch_add(1, butil::memory_order_relaxed); 
    int msg_index = thread_index; 
    std::deque<int64_t> timeq; 
    size_t MAX_QUEUE_SIZE = (size_t)req_rate; 
    if (MAX_QUEUE_SIZE < 100) { 
        MAX_QUEUE_SIZE = 100; 
    } else if (MAX_QUEUE_SIZE > 2000) { 
        MAX_QUEUE_SIZE = 2000; 
    } 
    timeq.push_back(butil::gettimeofday_us()); 
    while (!_stop) { 
        brpc::Controller* cntl = new brpc::Controller; 
        msg_index = (msg_index + _options.test_thread_num) % _msgs.size(); 
        Message* request = _msgs[msg_index]; 
        Message* response = _pbrpc_client->get_output_message(); 
        const int64_t start_time = butil::gettimeofday_us(); 
        google::protobuf::Closure* done = brpc::NewCallback< 
            RpcPress,  
            RpcPress*,  
            brpc::Controller*,  
            Message*,  
            Message*, int64_t> 
            (this, &RpcPress::handle_response, cntl, request, response, start_time); 
        const brpc::CallId cid1 = cntl->call_id(); 
        _pbrpc_client->call_method(cntl, request, response, done); 
        _sent_count << 1; 
 
        if (_options.test_req_rate <= 0) {  
            brpc::Join(cid1); 
        } else { 
            int64_t end_time = butil::gettimeofday_us(); 
            int64_t expected_elp = 0; 
            int64_t actual_elp = 0; 
            timeq.push_back(end_time); 
            if (timeq.size() > MAX_QUEUE_SIZE) { 
                actual_elp = end_time - timeq.front(); 
                timeq.pop_front(); 
                expected_elp = (int64_t)(1000000 * timeq.size() / req_rate); 
            } else { 
                actual_elp = end_time - timeq.front(); 
                expected_elp = (int64_t)(1000000 * (timeq.size() - 1) / req_rate); 
            } 
            if (actual_elp < expected_elp) { 
                usleep(expected_elp - actual_elp); 
            } 
        } 
    } 
}