1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
[English version](/docs/en/docs-06-workflow.md)
## 06 - 与workflow异步框架的结合
### 1. Server
下面我们通过一个具体例子来呈现
- Echo RPC在接收到请求时，向下游发起一次http请求
- 对下游请求完成后，我们将http response的body信息填充到response的message里，回复给客户端
- 我们不希望阻塞/占据着Handler的线程，所以对下游的请求一定是一次异步请求
- 首先，我们通过Workflow框架的工厂``WFTaskFactory::create_http_task``创建一个异步任务http_task
- 然后，我们利用RPCContext的``ctx->get_series()``获取到ServerTask所在的SeriesWork
- 最后，我们使用SeriesWork的``push_back``接口将http_task放到SeriesWork的后面
~~~cpp
class ExampleServiceImpl : public Example::Service
{
public:
void Echo(EchoRequest *request, EchoResponse *response, RPCContext *ctx) override
{
auto *http_task = WFTaskFactory::create_http_task("https://www.sogou.com", 0, 0,
[request, response](WFHttpTask *task) {
if (task->get_state() == WFT_STATE_SUCCESS)
{
const void *data;
size_t len;
task->get_resp()->get_parsed_body(&data, &len);
response->mutable_message()->assign((const char *)data, len);
}
else
response->set_message("Error: " + std::to_string(task->get_error()));
printf("Server Echo()\nget_req:\n%s\nset_resp:\n%s\n",
request->DebugString().c_str(),
response->DebugString().c_str());
});
ctx->get_series()->push_back(http_task);
}
};
~~~
### 2. Client
下面我们通过一个具体例子来呈现
- 我们并行发出两个请求，1个是rpc请求，1个是http请求
- 两个请求都结束后，我们再发起一次计算任务，计算两个数的平方和
- 首先，我们通过RPC Client的``create_Echo_task``创建一个rpc异步请求的网络任务rpc_task
- 然后，我们通过Workflow框架的工厂``WFTaskFactory::create_http_task``和``WFTaskFactory::create_go_task``分别创建异步网络任务http_task，和异步计算任务calc_task
- 最后，我们利用串并联流程图，乘号代表并行、大于号代表串行，将3个异步任务组合起来执行start
~~~cpp
void calc(int x, int y)
{
int z = x * x + y * y;
printf("calc result: %d\n", z);
}
int main()
{
Example::SRPCClient client("127.0.0.1", 1412);
auto *rpc_task = client.create_Echo_task([](EchoResponse *response, RPCContext *ctx) {
if (ctx->success())
printf("%s\n", response->DebugString().c_str());
else
printf("status[%d] error[%d] errmsg:%s\n",
ctx->get_status_code(), ctx->get_error(), ctx->get_errmsg());
});
auto *http_task = WFTaskFactory::create_http_task("https://www.sogou.com", 0, 0, [](WFHttpTask *task) {
if (task->get_state() == WFT_STATE_SUCCESS)
{
std::string body;
const void *data;
size_t len;
task->get_resp()->get_parsed_body(&data, &len);
body.assign((const char *)data, len);
printf("%s\n\n", body.c_str());
}
else
printf("Http request fail\n\n");
});
auto *calc_task = WFTaskFactory::create_go_task(calc, 3, 4);
EchoRequest req;
req.set_message("Hello!");
req.set_name("1412");
rpc_task->serialize_input(&req);
WFFacilities::WaitGroup wait_group(1);
SeriesWork *series = Workflow::create_series_work(http_task, [&wait_group](const SeriesWork *) {
wait_group.done();
});
series->push_back(rpc_task);
series->push_back(calc_task);
series->start();
wait_group.wait();
return 0;
}
~~~
### 3. Upstream
SRPC可以直接使用Workflow的任何组件，最常用的就是[Upstream](https://github.com/sogou/workflow/blob/master/docs/about-upstream.md)，SRPC的任何一种client都可以使用Upstream。
我们通过参数来看看如何构造可以使用Upstream的client：
```cpp
#include "workflow/UpstreamManager.h"
int main()
{
// 1. 创建upstream并添加实例
UpstreamManager::upstream_create_weighted_random("echo_server", true);
UpstreamManager::upstream_add_server("echo_server", "127.0.0.1:1412");
UpstreamManager::upstream_add_server("echo_server", "192.168.10.10");
UpstreamManager::upstream_add_server("echo_server", "internal.host.com");
// 2. 构造参数，填上upstream的名字
RPCClientParams client_params = RPC_CLIENT_PARAMS_DEFAULT;
client_params.host = "echo_server";
client_params.port = 1412; // 这个port只用于upstream URI解析，不影响具体实例的选取
// 3. 用参数创建client，其他用法与示例类似
Example::SRPCClient client(&client_params);
...
```
如果使用了ConsistentHash或者Manual方式创建upstream，则我们往往需要对不同的task进行区分、以供选取算法使用。这时候可以使用client task上的`int set_uri_fragment(const std::string& fragment);`接口，设置请求级相关的信息。
这个域的是URI里的fragment，语义请参考[RFC3689 3.5-Fragment](https://datatracker.ietf.org/doc/html/rfc3986#section-3.5)，任何需要用到fragment的功能（如其他选取策略里附带的其他信息），都可以利用这个域。
|
