后端技术_grpc程序编写
同步
proto文件
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// IM.Login.proto
syntax = "proto3";
package IM.Login;
// 定义服务
service ImLogin{
rpc Regist( IMRegistReq ) returns ( IMRegistRes ) {}
rpc Login( IMLoginReq ) returns ( IMLoginRes ) {}
}
// 注册账号
message IMRegistReq{
string user_name = 1;
string password = 2;
}
// 注册返回
message IMRegistRes{
string user_name = 1;
uint32 user_id = 2;
uint32 result_code = 3;
}
// 登陆账号
message IMLoginReq{
string user_name = 1;
string password = 2;
}
// 登录返回
message IMLoginRes{
uint32 user_id = 2;
uint32 result_code = 3;
}server端
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// server.cc
#
#
// grpc头文件
#
#
#
#
#
// grpc标准命名空间
using grpc::Server;
using grpc::ServerBuilder;
using grpc::ServerContext;
using grpc::Status;
// protobuf命名空间
using IM::Login::ImLogin;
using IM::Login::ImRegistReq;
using IM::Login::ImRegistRes;
using IM::Login::ImLoginReq;
using IM::Login::ImLoginRes;
// 继承服务
class IMLoginSeviceImpl : public ImLogin::Service {
// 注册方法
virtual Status Regist(ServerContext* context, const IMRegistReq* request, IMRegistReq* response) override {
response->set_user_name(request->user_name());
response->set_user_id(10);
response->set_result_code(0);
return Status::OK;
}
// 登陆方法
virtual Status Login(ServerContext* context, const IMLoginReq* request, IMLoginReq* response) override {
response->set_user_id(10);
response->set_result_code(0);
return Status::OK;
}
};
void RunServer()
{
std::string server_addr("0.0.0.0:50051");
IMLoginSeviceImpl service;
ServerBuilder builder;
builder.AddListeningPort(server_addr, grpc::InsecureServerCredentials());
builder.AddChannelArgument(GRPC_ARG_KEEPLIVE_TIME_MS, 5000);
builder.AddChannelArgument(GRPC_ARG_KEEPLIVE_TIMEOUT_MS, 10000);
builder.AddChannelArgument(GRPC_ARG_KEEPLIVE_PERMIT_WITHOUT_CALLS, 1);
builder.RegisterService(&service);
// 创建、启动
std::unique_ptr<Server> server(builder.BuildAndStart());
std::cout << "Server listening on " << server_addr << std::endl;
// 进入服务循环
server->Wait();
}
int main(int argc, const char** argv)
{
RunServer();
}client端
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// client.cc
#
#
// grpc头文件
#
#
#
#
#
// grpc标准命名空间
using grpc::Channel;
using grpc::ClientContext;
using grpc::Status;
// protobuf命名空间
using IM::Login::ImLogin;
using IM::Login::ImRegistReq;
using IM::Login::ImRegistRes;
using IM::Login::ImLoginReq;
using IM::Login::ImLoginRes;
class ImLoginClient
{
public:
ImLoginClient(std::shared_ptr<Channel> channel)
:stub_(ImLogin::NewStub(channel)){
}
void Regist(const std::string &user_name, const std::string &password)
{
IMRegistReq request;
request.set_user_name(user_name);
request.set_password(password);
IMRegistRes response;
ClientContext context;
Status status = stub_->Regist(&context, request, &response);
if(status.ok()){
std::cout << "user_name:" << response.user_name() << ", user_id:" << response.user_id() << std::endl ;
} else {
std::cout << "user_name:" << response.user_name() << " Regist failed : " << response.result_code() << std::endl ;
}
}
void Login(const std::string &user_name, const std::string &password)
{
IMLoginReq request;
request.set_user_name(user_name);
request.set_password(password);
IMLoginRes response;
ClientContext context;
Status status = stub_->Regist(&context, request, &response);
if(status.ok()){
std::cout << "user_id:" << response.user_id() << " login ok" << std::endl ;
} else {
std::cout << "user_id:" << response.user_id() << " Login failed : " << response.result_code() << std::endl ;
}
}
private:
std::unique_ptr<ImLogin::Stub> stub_;
};
int main()
{
// 服务器的地址
std::string server_addr = "localhost:50051";
ImLoginClient im_login_client(
grpc::CreateChannel(server_addr, grpc::InsecureServerCredentials())
);
std::string user_name = "k";
std::string password = "123456";
im_login_client.Regist(user_name, password);
im_login_client.Login(user_name, password);
return 0;
}
异步
- 服务端
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
#
#
#
#
#
#
#
#
using grpc::Server;
using grpc::ServerAsyncResponseWriter;
using grpc::ServerBuilder;
using grpc::ServerCompletionQueue;
using grpc::ServerContext;
using grpc::Status;
// 自己proto文件的命名空间
using IM::Login::ImLogin;
using IM::Login::IMLoginReq;
using IM::Login::IMLoginRes;
using IM::Login::IMRegistReq;
using IM::Login::IMRegistRes;
class ServerImpl final{
public:
~ServerImpl(){
server_->Shutdown();
cq_->Shutdown();
}
void Run(){
std::string server_address("0.0.0.0:50051");
ServerBuilder builder;
builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
builder.RegisterService(&service_);
cq_ = builder.AddCompletionQueue();
server_ = builder.BuildAndStart();
std::cout << "Server listening on " << server_address << std::endl;
HandleRpcs();
}
private:
class CallDate{
public:
CallData(ImLogin::AsyncService* service, ServerCompletionQueue* cq):service_(service),cq_(cq),status_(CREATE){
Proceed();
}
virtual ~CallData(){}
virtual void Proceed() {return;}
ImLogin::AsyncService* service_;
ServerCompletionQueue* cq_;
ServerContext ctx_;
enum CallStatus { CREATE, PROCESS, FINISH };
CallStatus status_;
};
class RegistCallData : public CallData{
public:
RegistCallData(ImLogin::AsyncService* service, ServerCompletionQueue* cq)
:CallData(service, cq), responder_(&ctx_) {
Proceed();
}
~RegistCallData() {}
void Proceed() override {
if(status_ == CREATE){
status_ = PROCESS;
service_->RequestRegist(&ctx_, &request_, &responder_, cq_, cq_, this);
}else if(status_ == PROCESS){
new RegistCallData(service_, cq_);
reply_.set_user_name(request_.user_name());
reply_.set_user_id(10);
reply_.set_result_code(0);
status_ = FINISH;
responder_.Finish(reply_, Status::OK, this);
}else{
GPR_ASSERT(status_ == FINISH);
delete this;
}
}
private:
IMRegistReq request_;
IMRegistRes reply_;
ServerAsyncResponseWriter<IMRegistRes> responder_;
};
class LoginCallData : public CallData{
public:
LoginCallData(ImLogin::AsyncService* service, ServerCompletionQueue* cq)
:CallData(service, cq), responder_(&ctx_) {
Proceed();
}
~LoginCallData() {}
void Proceed() override {
if (status_ == CREATE) { // 0
status_ = PROCESS;
service_->RequestLogin(&ctx_, &request_, &responder_, cq_, cq_, this);
} else if (status_ == PROCESS) { // 1
new LoginCallData(service_, cq_); // 1. 创建处理逻辑
reply_.set_user_id(10);
reply_.set_result_code(0);
status_ = FINISH;
responder_.Finish(reply_, Status::OK, this);
} else {
GPR_ASSERT(status_ == FINISH);
delete this;
}
}
private:
IMLoginReq request_;
IMLoginRes reply_;
ServerAsyncResponseWriter<IMLoginRes> responder_;
};
void HandleRpcs()
{
new RegistCallData(&service_, cq_.get());
new LoginCallData(&service_, cq_.get());
void *tag;
bool ok;
while(true)
{
GPR_ASSERT(cq_->Next(&tag, &ok));
GPR_ASSERT(ok);
static_cast<CallData*>(tag)->Proceed();
}
}
std::unique_ptr<ServerCompletionQueue> cq_;
ImLogin::AsyncService service_;
std::unique_ptr<Server> server_;
}
int main(int argc, char** argv)
{
ServerImpl server;
server.Run();
return 0;
}