brpc学习：ParallelChannel的使用 (转）

1、为什么要有parallel channel
关于ParallelChannel的使用，首先需要知道parallelchannel实际是为了更好的并发编程使用brpc而提供的一套api。在之前的博客中已经提到了如何使用brpc完成同步、异步的并发操作。虽然它们能完成异步并发的操作，但是这类代码的多线程陷阱很多，用户可能写出了bug也不自知，复现和调试也比较困难。而且实现要么只能支持同步的情况，要么得为异步重写一套。以"在多个异步RPC完成后运行一些代码"为例，它的同步实现一般是异步地发起多个RPC，然后逐个等待各自完成。

它的异步实现一般是用一个带计数器的回调，每当一个RPC完成时计数器减一，直到0时调用回调。可以看到它的缺点：

1.同步和异步代码不一致。用户无法轻易地从一个模式转为另一种模式。从设计的角度，不一致暗示了没有抓住本质。
2.往往不能被取消。正确及时地取消一个操作不是一件易事，何况是组合访问。但取消对于终结无意义的等待是很必要的。
3.不能继续组合。比如你很难把一个上述实现变成“更大"的访问模式的一部分。换个场景还得重写一套。

我们需要更好的抽象。如果我们能以不同的方式把一些Channel组合为更大的Channel，并把不同的访问模式置入其中，那么用户可以便用统一接口完成同步、异步、取消等操作。这种channel在brpc中被称为组合channel。

2、什么是parallel channel
ParallelChannel (有时被称为“pchan”)同时访问其包含的sub channel，并合并它们的结果。用户可通过CallMapper修改请求，通过ResponseMerger合并结果。ParallelChannel看起来就像是一个Channel：

支持同步和异步访问。
发起异步操作后可以立刻删除。
可以取消。
支持超时。
一个sub channel可多次加入同一个ParallelChannel。当你需要对同一个服务发起多次异步访问并等待它们完成的话，这很有用。

ParallelChannel的内部结构大致如下：

从这个结构图可以看到，实际parallel channel是组合了多个sub channel，然后将请求request发送到对应的所有sub channel上，然后将每个子response的结果，最后做一个合并传输给用户。

总的来说，用户自己去写多线程的同步异步操作可能会有很多bug，这里使用parallelchannel使用就不用担心这个问题，用户只需要记住加入哪几个sub channel，所有的线程接入的接口就是parallel channel，不用是关乎线程的竞争问题。sub channel可以在内部保证正确的线程rpc调用。

3、bvar计数器
在brpc中使用了大量的bvar，bvar是brpc内部的监控器，它可以监控在parallel channel中各个sub channel的性能（qps,io延时等等），详细的bvar的介绍和说明可以参考这篇博客：brpc学习：bvar

常用的bvar有：
•bvar::Adder : 计数器，默认0，varname << N相当于varname += N。
•bvar::Maxer : 求最大值，默认std::numeric_limits::min()，varname << N相当于varname = max(varname, N)。
•bvar::Miner : 求最小值，默认std::numeric_limits::max()，varname << N相当于varname = min(varname, N)。
•bvar::IntRecorder : 求自使用以来的平均值。注意这里的定语不是“一段时间内”。一般要通过Window衍生出时间窗口内的平均值。
•bvar::Window : 获得某个bvar在一段时间内的累加值。Window衍生于已存在的bvar，会自动更新。
•bvar::PerSecond : 获得某个bvar在一段时间内平均每秒的累加值。PerSecond也是会自动更新的衍生变量。
•bvar::LatencyRecorder : 专用于记录延时和qps的变量。输入延时，平均延时/最大延时/qps/总次数 都有了。
•bvar::Status：记录和显示一个值，拥有额外的set_value函数。
•bvar::PassiveStatus：按需显示值。在一些场合中，我们无法setvalue或不知道以何种频率setvalue，更适合的方式也许是当需要显示时才打印。用户传入打印回调函数实现这个目的。

4.实例
这里给出一个例子，这个例子来自brpc开源的parallel channel例子。例子主要思想是，开启50个线程去调用远程的rpc服务。每个parallel channel中设置有3个sub channel。最后通过bvar监控每个sub channel的性能并打印。
代码如下：

#include <gflags/gflags.h>
#include <bthread/bthread.h>
#include <butil/logging.h>
#include <butil/string_printf.h>
#include <butil/time.h>
#include <butil/macros.h>
#include <brpc/parallel_channel.h>    //包含parallel_channel模块的头文件
#include <brpc/server.h>            
#include "echo.pb.h"                

DEFINE_int32(thread_num, 50, "Number of threads to send requests");                //使用线程的个数
DEFINE_int32(channel_num, 3, "Number of sub channels");                            //使用sub channel的个数
DEFINE_bool(same_channel, false, "Add the same sub channel multiple times");    //是不是添加同一个sub channel
DEFINE_bool(use_bthread, false, "Use bthread to send requests");                //是不是使用bthread 还是普通的pthread
DEFINE_int32(attachment_size, 0, "Carry so many byte attachment along with requests");
DEFINE_int32(request_size, 16, "Bytes of each request");
DEFINE_string(connection_type, "", "Connection type. Available values: single, pooled, short");
DEFINE_string(protocol, "baidu_std", "Protocol type. Defined in src/brpc/options.proto");
DEFINE_string(server, "0.0.0.0:8002", "IP Address of server");
DEFINE_string(load_balancer, "", "The algorithm for load balancing");
DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds");
DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)"); 
DEFINE_bool(dont_fail, false, "Print fatal when some call failed");
DEFINE_int32(dummy_port, -1, "Launch dummy server at this port");

std::string g_request;
std::string g_attachment;
bvar::LatencyRecorder g_latency_recorder("client");                                //注意，这里使用的是bvar来定义的，bvar是brpc内置的计数器，可以自行监控某个sub channel的io
bvar::Adder<int> g_error_count("client_error_count");                            //错误计数器
bvar::LatencyRecorder* g_sub_channel_latency = NULL;                            //延时计数器

//定义一个sender函数，这个函数的功能是完成rpc的同步调用，并且当rpc返回的时候，返回bvar监控的一些latency qps的数据，并写入定义的计数器中
static void* sender(void* arg) {
    // Normally, you should not call a Channel directly, but instead construct
    // a stub Service wrapping it. stub can be shared by all threads as well.
    example::EchoService_Stub stub(static_cast<google::protobuf::RpcChannel*>(arg));

    int log_id = 0;
    while (!brpc::IsAskedToQuit()) {
        // We will receive response synchronously, safe to put variables
        // on stack.
        example::EchoRequest request;
        example::EchoResponse response;
        brpc::Controller cntl;
        
        //有附件，则给服务端反射附件
        request.set_value(log_id++);
        if (!g_attachment.empty()) {
            // Set attachment which is wired to network directly instead of 
            // being serialized into protobuf messages.
            cntl.request_attachment().append(g_attachment);
        }
        
        //调用rpc
        // Because `done'(last parameter) is NULL, this function waits until
        // the response comes back or error occurs(including timedout).
        stub.Echo(&cntl, &request, &response, NULL);
        if (!cntl.Failed()) {
            //rpc调用成功之后
            //在controller中将整个channel的latecy和sun channel的值发送到bvar中保存
            g_latency_recorder << cntl.latency_us();
            for (int i = 0; i < cntl.sub_count(); ++i) {
                if (cntl.sub(i) && !cntl.sub(i)->Failed()) {
                    g_sub_channel_latency[i] << cntl.sub(i)->latency_us();
                }
            }
        } else {
            //rpc调用失败之后
            //将失败的次数信息写入到bvar中
            g_error_count << 1;
            CHECK(brpc::IsAskedToQuit() || !FLAGS_dont_fail)
                << "error=" << cntl.ErrorText() << " latency=" << cntl.latency_us();
            // We can't connect to the server, sleep a while. Notice that this
            // is a specific sleeping to prevent this thread from spinning too
            // fast. You should continue the business logic in a production 
            // server rather than sleeping.
            bthread_usleep(50000);
        }
    }
    return NULL;
}

int main(int argc, char* argv[]) {
    // Parse gflags. We recommend you to use gflags as well.
    GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true);

    // A Channel represents a communication line to a Server. Notice that 
    // Channel is thread-safe and can be shared by all threads in your program.
    // 设置一个parallel channel，并开启这个channel
    brpc::ParallelChannel channel;
    brpc::ParallelChannelOptions pchan_options;
    pchan_options.timeout_ms = FLAGS_timeout_ms;
    if (channel.Init(&pchan_options) != 0) {
        LOG(ERROR) << "Fail to init ParallelChannel";
        return -1;
    }
    
    //设置sub channel的相关属性
    brpc::ChannelOptions sub_options;
    sub_options.protocol = FLAGS_protocol;
    sub_options.connection_type = FLAGS_connection_type;
    sub_options.max_retry = FLAGS_max_retry;
    // Setting sub_options.timeout_ms does not work because timeout of sub 
    // channels are disabled in ParallelChannel.
    
    //通过gflags判断是否是同一个sub
    if (FLAGS_same_channel) {
        // For brpc >= 1.0.155.31351, a sub channel can be added into
        // a ParallelChannel more than once.
        
        //如果是同一个sub channel，只用new一次，之后就不用new了
        brpc::Channel* sub_channel = new brpc::Channel;
        // Initialize the channel, NULL means using default options. 
        // options, see `brpc/channel.h'.
        if (sub_channel->Init(FLAGS_server.c_str(), FLAGS_load_balancer.c_str(), &sub_options) != 0) {
            LOG(ERROR) << "Fail to initialize sub_channel";
            return -1;
        }
        
        //将sub channel添加进channel中
        for (int i = 0; i < FLAGS_channel_num; ++i) {
            if (channel.AddChannel(sub_channel, brpc::OWNS_CHANNEL,
                                   NULL, NULL) != 0) {
                LOG(ERROR) << "Fail to AddChannel, i=" << i;
                return -1;
            }
        }
    } else {    //通过gflags判断不是一个sub channel，这个时候就要每次new()一个sub channel
        for (int i = 0; i < FLAGS_channel_num; ++i) {
            brpc::Channel* sub_channel = new brpc::Channel;
            // Initialize the channel, NULL means using default options. 
            // options, see `brpc/channel.h'.
            if (sub_channel->Init(FLAGS_server.c_str(), FLAGS_load_balancer.c_str(), &sub_options) != 0) {
                LOG(ERROR) << "Fail to initialize sub_channel[" << i << "]";
                return -1;
            }
            if (channel.AddChannel(sub_channel, brpc::OWNS_CHANNEL,
                                   NULL, NULL) != 0) {
                LOG(ERROR) << "Fail to AddChannel, i=" << i;
                return -1;
            }
        }
    }

    // Initialize bvar for sub channel
    // 初始化bvar,这里定义了LatencyRecorder的数组，用来记录每个sub channel的返回记录信息
    g_sub_channel_latency = new bvar::LatencyRecorder[FLAGS_channel_num];
    for (int i = 0; i < FLAGS_channel_num; ++i) {
        std::string name;
        butil::string_printf(&name, "client_sub_%d", i);
        g_sub_channel_latency[i].expose(name);
    }

    if (FLAGS_attachment_size > 0) {
        g_attachment.resize(FLAGS_attachment_size, 'a');
    }
    if (FLAGS_request_size <= 0) {
        LOG(ERROR) << "Bad request_size=" << FLAGS_request_size;
        return -1;
    }
    g_request.resize(FLAGS_request_size, 'r');

    if (FLAGS_dummy_port >= 0) {
        brpc::StartDummyServerAt(FLAGS_dummy_port);
    }

    //通过两种不同调用多线程的方式进行，一种是通过brpc推荐的bthread，一种是Linux下常见的pthread
    std::vector<bthread_t> bids;
    std::vector<pthread_t> pids;
    //通过gflags命令行来判断使用哪种方式
    if (!FLAGS_use_bthread) {
        pids.resize(FLAGS_thread_num);
        for (int i = 0; i < FLAGS_thread_num; ++i) {
            if (pthread_create(&pids[i], NULL, sender, &channel) != 0) {                //pthread创建线程的方式,每个线程的回调函数都是sender
                LOG(ERROR) << "Fail to create pthread";
                return -1;
            }
        }
    } else {
        bids.resize(FLAGS_thread_num);
        for (int i = 0; i < FLAGS_thread_num; ++i) {
            if (bthread_start_background(                                                //bthread的创建方式,每个线程的回调函数都是sender
                    &bids[i], NULL, sender, &channel) != 0) {
                LOG(ERROR) << "Fail to create bthread";
                return -1;
            }
        }
    }
    
    //当每个线程同步接收到消息的时候，将返回的消息保存在bvar中
    while (!brpc::IsAskedToQuit()) {
        sleep(1);
        //打印总的延迟信息，同时打印各个sun channel的信息
        LOG(INFO) << "Sending EchoRequest at qps=" << g_latency_recorder.qps(1)
                  << " latency=" << g_latency_recorder.latency(1) << noflush;
        for (int i = 0; i < FLAGS_channel_num; ++i) {
            LOG(INFO) << " latency_" << i << "=" 
                      << g_sub_channel_latency[i].latency(1)
                      << noflush;
        }
        LOG(INFO);
    }
    
    //关闭所有的线程
    LOG(INFO) << "EchoClient is going to quit";
    for (int i = 0; i < FLAGS_thread_num; ++i) {
        if (!FLAGS_use_bthread) {
            pthread_join(pids[i], NULL);                                            //join等待线程结束后回收对应的线程资源
        } else {
            bthread_join(bids[i], NULL);
        }
    }

    return 0;
}

5、测试结果
使用50个线程，3个sub_channel的测试结果：

使用3个线程，3个sub channel的测试结果：

使用50个线程，10个sub channel的测试结果：

从上面的测试结果可以看到以下几个小节：

sub channel的增加肯定是能增大qps的，如果把client当做channel，那么线程就是访问这个client，client越多，相对应的连接带宽肯定是变大的。
开启的线程去访问某个服务，不用去关系线程之间的同步互斥问题。

原文链接：https://blog.csdn.net/u012414189/article/details/84634495