workflow后台服务框架使用记录--自定义服务端/客户端
作者：互联网
Workflow是Sogou研发的一款性能优异的C++网络服务框架，纯异步+并行的特性使它拥有极高的性能，核心优势：
1.多线程服务下的阻塞操作；
2.网络线程和执行线程之间的调度策略；
3.底层I/O方式的高效封装epoll；
我们可以workflow提供的底层协议与API，轻松实现自定义十万级并发高性能服务端/客户端。首先make编译workflow，注意必须使用cmake3以上版本进行编译：
服务端工作模式如下：
新建服务端：
进入/workflow_server/workflow/tutorial目录，并建立如下文件：
1 #include <stdio.h>
2 #include "workflow/WFHttpServer.h"
3
4 int main()
5 {
6 WFHttpServer server([](WFHttpTask *task) {
7 task->get_resp()->append_output_body("<html>Hello World!</html>");
8 });
9
10 if (server.start(8888) == 0) { // start server on port 8888
11 getchar(); // press "Enter" to end.
12 server.stop();
13 }
14
15 return 0;
16 }
对于HTTP协议，我们构造对应对象是WFHttpServer，任务类型是WFHttpTask；server的消息处理方法是task->get_resq()，并填充回复字段，启动命令：
1 curl -i IP
新建客户端，使用命令行来指定默认版本or自定义版本，进入：
1 #include <string.h>
2 #include <stdio.h>
3 #include "workflow/Workflow.h"
4 #include "workflow/HttpMessage.h"
5 #include "workflow/WFTaskFactory.h"
6 #include "workflow/WFFacilities.h"
7 #include "message.h"
8
9 using WFTutorialTask = WFNetworkTask<protocol::TutorialRequest,
10 protocol::TutorialResponse>;
11 using tutorial_callback_t = std::function<void (WFTutorialTask *)>;
12
13 using namespace protocol;
14
15 class MyFactory : public WFTaskFactory
16 {
17 public:
18 static WFTutorialTask *create_tutorial_task(const std::string& host,
19 unsigned short port,
20 int retry_max,
21 tutorial_callback_t callback)
22 {
23 using NTF = WFNetworkTaskFactory<TutorialRequest, TutorialResponse>;
24 WFTutorialTask *task = NTF::create_client_task(TT_TCP, host, port,
25 retry_max,
26 std::move(callback));
27 task->set_keep_alive(30 * 1000);
28 return task;
29 }
30 };
31
32 int main(int argc, char *argv[])
33 {
34 unsigned short port;
35 std::string host;
36 std::string self_defined;
37
38 if (argc != 4)
39 {
40 fprintf(stderr, "USAGE: %s <host> <port> <selection>\n", argv[0]);
41 exit(1);
42 }
43
44 host = argv[1];
45 port = atoi(argv[2]);
46 self_defined = argv[3]; //选项参数
47
48 if(self_defined == "default"){
49 std::function<void (WFTutorialTask *task)> callback =[&host, port, &callback](WFTutorialTask *task) {
50 int state = task->get_state();
51 int error = task->get_error();
52 TutorialResponse *resp = task->get_resp();
53 char buf[1024];
54 void *body;
55 size_t body_size;
56
57 if (state != WFT_STATE_SUCCESS)
58 {
59 if (state == WFT_STATE_SYS_ERROR)
60 fprintf(stderr, "SYS error: %s\n", strerror(error));
61 else if (state == WFT_STATE_DNS_ERROR)
62 fprintf(stderr, "DNS error: %s\n", gai_strerror(error));
63 else
64 fprintf(stderr, "other error.\n");
65 return;
66 }
67
68 resp->get_message_body_nocopy(&body, &body_size);
69 if (body_size != 0)
70 printf("Server Response: %.*s\n", (int)body_size, (char *)body);
71
72 printf("Input next request string (Ctrl-D to exit): ");
73 *buf = '\0';
74 scanf("%1023s", buf);
75 body_size = strlen(buf);
76 if (body_size > 0)
77 {
78 WFTutorialTask *next;
79 next = MyFactory::create_tutorial_task(host, port, 0, callback);
80 next->get_req()->set_message_body(buf, body_size);
81 next->get_resp()->set_size_limit(4 * 1024);
82 **task << next; /* equal to: series_of(task)->push_back(next) */
83 }
84 else
85 printf("\n");
86 };
87
88 /* First request is emtpy. We will ignore the server response. */
89 WFFacilities::WaitGroup wait_group(1);
90 WFTutorialTask *task = MyFactory::create_tutorial_task(host, port, 0, callback);
91 task->get_resp()->set_size_limit(4 * 1024);
92 Workflow::start_series_work(task, [&wait_group](const SeriesWork *) {
93 wait_group.done();
94 });
95
96 wait_group.wait();
97 }
98 else
99 {
100 const char *url = "https://www.baidu.com/"; //指定IP
101 WFHttpTask *task = WFTaskFactory::create_http_task (url, 2, 3,
102 [](WFHttpTask * task) {
103 fprintf(stderr, "%s %s %s\r\n",
104 task->get_resp()->get_http_version(),
105 task->get_resp()->get_status_code(),
106 task->get_resp()->get_reason_phrase());
107 });
108 task->start();
109 getchar(); // press "Enter" to end.
110 }
111 return 0;
112 }
对于客户端性能的提升，除了异步回调机制来取回http任务响应之外，还有客户端连接的复用，即长连接池来避免频繁新建TCP连接和握手降低处理效率：
启动指令如下：
1 ./client host_ip port selection
QPS压测工具：
本次实验我们使用的压测工具为wrk和wrk2。 前者适合测试特定并发下的QPS极限和延时， 后者适合在特定QPS下测试延时分布。
我们也尝试过使用其他测试工具，例如ab等，但无法打出足够的压力。还可以参考基于Sogou C++ Workflow的benchmark工具。
编译：
进入/wrk目录，执行make编译wrk工具；
压测指令：
首先启动http_server
1 ./wrk --latency -d10 -c200 --timeout 8 -t 6 http://127.0.0.1:9000
命令行解释
-c200: 启动200个连接
-t6: 开启6个线程做压力测试
-d10: 压测持续10s
--timeout 8: 连接超时时间8s
对于本次测试的阿里云ECS单核服务器，难以达到多核并行线程处理模式下的数十万QPS；详细性能数据可以参考：workflow/README.md at master · DeshZhao/workflow (github.com)；更多应用参考：FAQ（持续更新） · Issue #170 · sogou/workflow (github.com)
总结：workflow框架对于Http任务的编排有超高的灵活性，对于十万级以上的QPS，并发性能远好于Nginx/brpc等框架。
标签：body,task,自定义,get,--,workflow,include,size 来源： https://www.cnblogs.com/Desh/p/16279087.html