使用gPRC提供服务#

1
syntax = "proto3";
2

3
package message;
4

5
// 定义消息类型
6
message GetChatServerReq {
7
  int32 uid = 1;
8
}
9

10

11
message GetChatServerRsp {
12
  int32 error = 1;
13
  string host = 2;
14
  string port = 3;
15
  string token = 4;
16
}
17

18
// 定义服务的接口
19
service StatusService {
20
  rpc GetChatServer (GetChatServerReq) returns (GetChatServerRsp) {}
21
}

1
void RunServer() {
2
  auto& cfg = ConfigManager::GetInstance();
3

4
  // gRPC监听的地址
5
  std::string server_address(cfg["StatusServer"]["Host"] + ":" + cfg["StatusServer"]["Port"]);
6

7
  // gRPC服务实例
8
  StatusServiceImpl service;
9

10
  // 创建gRPC服务器
11
  gRPC::ServerBuilder builder;
12
  // 监听端口和添加服务
13
  builder.AddListeningPort(server_address, gRPC::InsecureServerCredentials());
14
  builder.RegisterService(&service);
15

16
  // 构建并启动gRPC服务器
17
  std::unique_ptr<gRPC::Server> server(builder.BuildAndStart());
18
  std::cout << "Server listening on " << server_address << std::endl;
19

20
  // 创建Boost.Asio的io_context监听终止信号
21
  boost::asio::io_context io_context;
22
  boost::asio::signal_set signals(io_context, SIGINT, SIGTERM);
23
  signals.async_wait([&server](const boost::system::error_code& error, int signal_number) {
24
    if (!error) {
25
      std::cout << "Shutting down server..." << std::endl;
26
      server->Shutdown(); // 优雅地关闭服务器
27
    }
28
    });
29

30
  // 守护线程运行io_context，捕获SIGINT信号
31
  std::thread([&io_context]() { io_context.run(); }).detach();
32

33
  // 等待gRPC服务器关闭
34
  server->Wait();
35
  // 停止boost的io_context服务
36
  io_context.stop();
37
}

1
#pragma once
2
#include <gRPCpp/gRPCpp.h>
3
#include "message.gRPC.pb.h"
4

5
using gRPC::Server;
6
using gRPC::ServerBuilder;
7
using gRPC::ServerContext;
8
using gRPC::Status;
9
using message::GetChatServerReq;
10
using message::GetChatServerRsp;
11
using message::StatusService;
12

13
// ChatServer类，用地址和端口表示
14
struct ChatServer {
15
  std::string host;
16
  std::string port;
17
};
18

19
class StatusServiceImpl final : public StatusService::Service
20
{
21
public:
22
  StatusServiceImpl();
23
  Status GetChatServer(ServerContext* context, const GetChatServerReq* request,
24
    GetChatServerRsp* response) override;
25

26
  // 聊天服务器列表
27
  std::vector<ChatServer> _servers;
28
  // 指定下一个新聊天的服务器，简单的负载均衡
29
  int _server_index;
30
};

1
#include "StatusServiceImpl.h"
2
#include "ConfigManager.h"
3
#include "const.h"
4
std::string generate_unique_string() {
5
  // 创建UUID对象
6
  boost::uuids::uuid uuid = boost::uuids::random_generator()();
7

8
  // 将UUID转换为字符串
9
  std::string unique_string = to_string(uuid);
10

11
  return unique_string;
12
}
13

14
Status StatusServiceImpl::GetChatServer(ServerContext* context, const GetChatServerReq* request, GetChatServerRsp* response)
15
{
16
  std::string prefix("status server has received :  ");
17
  // 这里数组越界
18
  //_server_index = (_server_index++) % (_servers.size());
19
  // 轮询方式选择聊天服务器。
20
  _server_index = (_server_index + 1) % (_servers.size());
21
  auto& server = _servers[_server_index];
22

23
  // 设置返回的聊天服务器的地址和端口
24
  // 设置状态码，和访问令牌
25
  response->set_host(server.host);
26
  response->set_port(server.port);
27
  response->set_error(ErrorCodes::SUCCESS);
28
  response->set_token(generate_unique_string());
29
  std::cout << "Sending grcp response: " << response->DebugString() << std::endl;
30
  return Status::OK;
31
}
32

33
StatusServiceImpl::StatusServiceImpl() :_server_index(0)
34
{
35
  auto& cfg = ConfigManager::GetInstance();
36
  ChatServer server;
37
  // 初始化可用的聊天服务器列表
38
  server.port = cfg["ChatServer1"]["Port"];
39
  server.host = cfg["ChatServer1"]["Host"];
40
  _servers.push_back(server);
41

42
  server.port = cfg["ChatServer2"]["Port"];
43
  server.host = cfg["ChatServer2"]["Host"];
44
  _servers.push_back(server);
45
}