BRPC兼容GRPC stream

**Is your feature request related to a problem? (你需要的功能是否与某个问题有关?)**

  stream是rpc框架使用中的常用功能，虽然brpc有streaming rpc，但是兼容grpc stream能给框架的这部分功能带来更大的泛用性。
  经过百度内部Service Mesh实践，出于定制化需求方向的考虑，我们希望在proxyless模式下，brpc能够直连istio，逐渐减弱对envoy的依赖。istio下发配置使用的是双向grpc stream，所以我们需要完成brpc兼容grpc stream的适配。
 主要需求如下：
  1、兼容grpc steam的同步/异步或订阅推送式的客户端；
  2、一如既往简单便捷的客户端API，不希望定制protobuf插件。
 后续事项：*服务端API设计、直达底层h2协议的性能调优

**Describe the solution you'd like (描述你期望的解决方法)**

以grpc stream Demo的消息格式

为例，可以通过以下API来对grpc stream服务端进行访问。
```cpp
#include
#include
#include
#include
#include
#include

#include "grpcdemo.pb.h"

DEFINE_string(server, "0.0.0.0:8000", "IP Address of server");
DEFINE_string(protocol, "h2:grpc", "Protocol type. Defined in protocol/baidu/rpc/options.proto");
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_int32(interval_ms, 1000, "Milliseconds between consecutive requests");

using namespace brpc;

int main(int argc, char* argv[]) {
GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true);

StreamChannel channel;

StreamOptions options;
options.protocol = FLAGS_protocol;
options.timeout_ms = FLAGS_timeout_ms/*milliseconds*/;
options.max_retry = FLAGS_max_retry;
if (channel.Init(FLAGS_server.c_str(), FLAGS_load_balancer.c_str(), &options) != 0) {
LOG(ERROR) << "Fail to initialize channel";
return -1;
}

demo::GRPCDemo_Stub stub(&channel);

// 以下参照了grpc sync_stream的设计

{
// 客户端流模式（在一次调用中, 客户端可以多次向服务器传输数据, 但是服务器只能返回一次响应）
// rpc ClientStreamingMethod (stream Request) returns (Response);
demo::Request req;
demo::Response res;
Controller cntl;
auto writer = CreateStreamVisitor(
&demo::GRPCDemo_Stub::ClientStreamingMethod, &stub, &res, &cntl);

req.set_request_data("Hello!");
writer->Write(req);

req.set_request_data("World!");
writer->Write(req);
writer->WritesDone();

writer->Finish();
if (!cntl.Failed()) {
LOG(INFO) << "ClientStreamingMethod wrote req, res=" << res.response_data();
} else {
LOG(WARNING) << "ClientStreamingMethod failed, error=" << cntl.ErrorText();
}
}

{
// 服务端流模式（在一次调用中, 客户端只能一次向服务器传输数据, 但是服务器可以多次返回响应）
// rpc ServerStreamingMethod (Request) returns (stream Response);
demo::Request req;
demo::Response res;
brpc::Controller cntl;
auto reader = CreateStreamVisitor(
&demo::GRPCDemo_Stub::ServerStreamingMethod, &stub, &req, &cntl);
while (reader->Read(&res)) {
LOG(INFO) << "ServerStreamingMethod read res, res=" << res.response_data();
}
reader->Finish();
if (cntl.Failed()) {
LOG(WARNING) << "ServerStreamingMethod failed, error=" << cntl.ErrorText();
}
}

{
// 双向流模式 (在一次调用中, 客户端和服务器都可以向对方多次收发数据)
// rpc BidirectionalStreamingMethod (stream Request) returns (stream Response);
demo::Request req;
demo::Response res;
brpc::Controller cntl;
auto stream = CreateStreamVisitor(
&demo::GRPCDemo_Stub::BidirectionalStreamingMethod, &stub, &cntl);
req.set_request_data("Hello!");
std::thread writer([stream]() {
for (int i = 0; i < 3; ++i) {
stream->Write(req);
}
// 全双工每个方向需要单独关闭
// 阻塞到以上所有写入操作完成，并关闭写入
// 业务最好等全部写完
stream->WritesDone();
if (cntl.Failed()) {
LOG(WARNING) << "BidirectionalStreamingMethod writing failed, error=" << cntl.ErrorText();
}
});

while (stream->Read(&res)) {
LOG(INFO) << "BidirectionalStreamingMethod read res, res=" << res.response_data();
}
writer.join();
// 阻塞到两端都写完读完，关闭该流
stream->Finish();
if (cntl.Failed()) {
LOG(WARNING) << "BidirectionalStreamingMethod failed, error=" << cntl.ErrorText();
}
}

LOG(INFO) << "StreamEchoClient is going to quit";
return 0;
}
```
具体设计
20220818更新设计概图
TBD

**Describe alternatives you've considered (描述你想到的折衷方案)**
简单地直接引入GRPC依赖

**Additional context/screenshots (更多上下文/截图)**

Start by reviewing the proposed client usage in brpc/stream_channel.h and brpc/grpc_stream.h, along with the linked grpc demo.proto format. The requested scope includes synchronous, asynchronous or subscription-style client support for client, server and bidirectional streams, while the service API and h2 performance work remain TBD. Done criteria are not fully defined in the issue.

Written by the indexing model from the issue text.