分布式缓存的核心就是gRPCgRPC的作用就是远程通信一个结点联系另外一个结点中间的桥梁就是gRPC。gRPC会携带交流沟通的数据我们当要请求一个数据的时候会携带缓存组的名字然后远程结点通过GetCacheGroup(name)来获取内部的缓存组。每一个结点都会有一个自己的gRPC为了管理这些gRPC我们创建了一个etcd注册器etcd注册器通过租约的机制保证服务的高效性具体就是每隔10s一个服务器重新向etcd发送消息证明自己状态正常并续约下一个10s这样子的好处是一旦有一个服务器断开了连接其他服务器可以很快的知道这个信息之后就不会向这个断开的服务器发送消息了。etcd就是中间联系人的这个角色所有结点向etcd注册自己同时也监听着其他结点的变化。gRPC每一个gRPC都需要一个配置信息表etcd通常采用集群式的部署所以我们用数组存储etcd服务端的地址。cert_file存储的式证书文件的路径用于提供身份凭据这个凭据一般都是公开的说明的是你是谁而key_file 是私密文件路径这个是不公开的证明你是谁。struct ServerOptions { std::vectorstd::string etcd_endpoints; // etcd 服务的地址列表 std::chrono::milliseconds dial_timeout; // 建立连接的超时时间 int max_msg_size; // 允许的最大消息大小 bool tls; // 是否启用 TLS用于加密通信 std::string cert_file; // TLS 证书文件的路径用于提供身份凭据 std::string key_file; // 与证书配套的私钥文件路径 ServerOptions() : etcd_endpoints({http://127.0.0.1:2379}) , dial_timeout(std::chrono::seconds(5)) , max_msg_size(4 20) // 4MB , tls(false) {} }; using ServerOption std::functionvoid(ServerOptions*);我们之后要修改server的配置信息这里提供了一个统一的配置操作统一我们的接口防止之后的代码杂乱。我们希望之后可以写成这样ServerOption a WithEtcdEndPoints(endpoints); a(server);先拿到函数对象然后把我们的目的服务器传入进去// 把 etcd 地址列表包装成一个配置操作之后用它修改 ServerOptions /* 使用方法是先赋值endpoints然后拿到相对应的函数对象再把我们的ServerOptions对象传入这个函数 */ inline auto WithEtcdEndPoints(const std::vectorstd::string endpoints) - ServerOption { return [endpoints](ServerOptions* o) {o-etcd_endpoints endpoints;}; } inline auto WithDialTimeout(std::chrono::milliseconds timeout) - ServerOption { return [timeout](ServerOptions* o) { o-dial_timeout timeout;}; } inline auto WithTLS(const std::string certfile, const std::string keyFile) - ServerOption { return [certfile, keyFile](ServerOptions* o) { o-tls true; o-cert_file certfile; o-key_file keyFile; }; }每一个服务器既是服务器也是客户端grpc就是一个数据传递的功能然后我们用proto来规定我们交流信息的格式一个缓存结点里面有一个grpc服务器一个etcd注册器这个相当于是和etcd沟通的桥梁里面需要包含注册函数和取消注册的函数还有续约的函数class CacheServer final : public pb::Cache::Service { public: CacheServer(const std::string addr, const std::string svc_name, ServerOptions opts ServerOptions{}); ~CacheServer() default; /* context 保存本次 RPC 调用的信息例如客户端地址、截止时间和取消状态 context-IsCancelled() request 客户端发来的请求这里包含要查询的缓存键 request-key() response 服务端填写的响应gRPC 会将它发送给客户端 response-set_value(缓存内容) */ auto Get(grpc::ServerContext* context, const pb::GetRequest* request, pb::GetResponse* response) - grpc::Status override; auto Set(grpc::ServerContext* context, const pb::SetRequest* request, pb::SetResponse* response) - grpc::Status override; auto Delete(grpc::ServerContext* context, const pb::GetRequest* request, pb::DeleteResponse* response) - grpc::Status override; auto Invalidate(grpc::ServerContext* context, const pb::GetRequest* request, pb::InvalidateResponse* response) - grpc::Status override; void Start(); void Stop(); private: auto LoadTLSCredentials(const std::string cert_file, const std::string key_file) - std::shared_ptrgrpc::ServerCredentials; std::string addr_; std::string svc_name_; std::unique_ptrgrpc::Server grpc_server_; std::unique_ptrEtcdRegistry etcd_register_; std::atomicbool is_stop_; ServerOptions opts_; };Etcd 注册服务器etcd注册类我们把服务器的地址和服务名写入的时候格式一定要统一一个etcdRegister里面需要有和etcd通信的etcd_client_本机在服务器里存储的形式然后我们续约这个操作不是在本地线程里面完成的因为续约这一个操作我们打算设计成一个while循环里面睡眠几秒之后醒来重新检查所以不适合把这个操作和注册这种快速的操作放在一起执行。class EtcdRegistry { public: EtcdRegistry(const std::string endpoints http://127.0.0.1:2379) { etcd_client_ std::make_uniqueetcd::Client(endpoints); } ~EtcdRegistry() default; // 将服务名和地址写入 etcd格式为 /services/{svc_name}/{addr},并绑定一个租约lease // 这样其他服务可以通过 etcd 查询到所有可用节点 bool Register(const std::string svc_name, std::string addr); // 撤销租约并删除服务信息确保结点下线的时候不会被其他服务继续发现 void Unregister(); private: auto GetLocalIP() - std::string; void KeepAliveLoop(); int64_t lease_id_{0}; std::unique_ptretcd::Client etcd_client_; // etcd_client_ 是负责与 etcd 通信的对象 std::string key_; // 服务器信息在etcd里面的存储形式 std::thread keepalive_thread_; std::atomicbool is_stop_{false}; };我们先把key_拼接出来然后通过桥梁etcd_client租一个10s的续约auto is_ok etcd_client_-put(key_, addr, lease_id_).get();这一步是核心关键我们把服务器的信息etcd的地址服务器的租约id传入进去保存到etcd服务器的数据里面之后如果断开连接etcd就会通过lease_id删除对应的数据。在注册了之后我们会执行续约线程bool EtcdRegistry::Register(const std::string svc_name, std::string addr) { std::string local_ip GetLocalIP(); if (local_ip.empty()) { return false; } // 如果地址只写了端口就在前面补上本机 IP if (!addr.empty() addr[0] :) { addr local_ip addr; } key_ /services/ svc_name / addr; // 创建租约 // 向 etcd 申请一个 TTL 为 10 秒的租约并等待申请结果这个操作是异步操作但是因为调用了get变成了阻塞等待 auto lease_resp etcd_client_-leasegrant(10).get(); if (!lease_resp.is_ok()) { return false; } // 从 etcd 的响应中取出租约 ID lease_id_ lease_resp.value().lease(); // 注册服务 auto is_ok etcd_client_-put(key_, addr, lease_id_).get(); if (!is_ok.is_ok()) { return false; } // 启动续约线程 keepalive_thread_ std::thread{[this] {this-KeepAliveLoop();}}; return true; } void EtcdRegistry::Unregister() { is_stop_ true; if (keepalive_thread_.joinable()) { keepalive_thread_.join(); } if (lease_id_ 0) { auto is_ok etcd_client_-leaserevoke(lease_id_).wait(); std::cout lease lease_id_ revoked successfully std::endl; } }这里涉及到了计算机网络的知识首先计算机对外有着一个独一无二的地址但是对内一般都是一个私有的地址比如127.168.189.133我们不需要这种无法对外面公开的地址因为外面通过这个地址没有办法找到对应的服务器所以我们需要获取一个Linux本机的网络接口信息。这里我们使用了一个ifaddrs结构体这个结构体调用getifaddrs()API可以获得本机所有的对外地址我们只需要一个就可以所以我们的操作是遍历ifa-afa_addr的所有地址当遇到了一个不是127.0.0.1的地址时本地地址我们利用网络编程的理论将其转化为sockaddr_in的形式这样子方便将二进制形式的IP转化为点分十进制。最后返回我们所获得的ip地址。// 得到一个外地可以访问的地址 auto EtcdRegistry::GetLocalIP() - std::string { /* 这行声明了一个指向 ifaddrs 结构体的指针通常用来获取 Linux 本机的网络接口信息。 它包含接口名称、地址以及指向下一条信息的指针等 ifa_name // 接口名例如 eth0、lo ifa_addr // 接口地址 ifa_next // 下一条记录形成链表 */ struct ifaddrs* ifaddr; // 因为 getifaddrs() 获取的就是本机网络接口的地址 // 所以遍历出来的地址本来就属于本机不是因为“非 127”才判断为本机地址 if (getifaddrs(ifaddr) -1) { return ; } std::string ip; for (struct ifaddrs* ifa ifaddr; ifa; ifa ifa-ifa_next) { if (!ifa-ifa_addr || ifa-ifa_addr-sa_family ! AF_INET) { continue; } char buf[INET_ADDRSTRLEN]; void* addr_ptr ((struct sockaddr_in*)ifa-ifa_addr)-sin_addr; // 这个信息是存储在网络里面的把二进制形式的 IP 地址转换成点分十进制 inet_ntop(AF_INET, addr_ptr, buf, INET_ADDRSTRLEN); std::string candidate{buf}; if (candidate ! 127.0.0.1) { ip candidate; break; } } freeifaddrs(ifaddr); return ip; }这个是续约的操作其实就是创建一个续约对象在check中不断地续约void EtcdRegistry::KeepAliveLoop() { while(!is_stop_) { etcd::KeepAlive keepalive{*etcd_client_, 10, lease_id_}; try { keepalive.Check(); } catch (const std::exception e) { std::cerr keepalive exception : e.what() std::endl; keepalive.Cancel(); } catch (...) { keepalive.Cancel(); } // 等待间隔 租约时间的1/3 std::this_thread::sleep_for(std::chrono::seconds(3)); } std::cout KeepAlive loop exited for lease {} lease_id_ std::endl; }所以总地来说,我们这里写地是etcd注册服务器而不是etcd的实现etcd实际上是一个分布式存储而我们这里写的是分布式缓存这里我们借助etcd分布式存储实现了一个服务注册的功能监视缓存结点是否有效。CacheServer每当创建一个缓存结点都需要在etcd里面注册相关的信息CacheServer::CacheServer(const std::string addr, const std::string svc_name, ServerOptions opts) : addr_(addr), svc_name_(svc_name), opts_(opts) { // 创建etcd住粗器传入的是etcd的地址 etcd_register_ std::make_uniqueEtcdRegistry(opts_.etcd_endpoints[0]); if (!etcd_register_-Register(svc_name, addr_)) { throw std::runtime_error([cache] failed to register service with etcd); } }对于数据的操作其实实现都是差不多的我们用户在传入请求的时候必须要附带着我们的请求组所以在本地结点查找缓存组找到缓存组之后通过key来找到我们需要的数据当我们取得数据之后就要返回给客户端auto CacheServer::Get(grpc::ServerContext* context, const pb::GetRequest* request, pb::GetResponse* response) - grpc::Status { // 先找到对应的组 auto group GetCacheGroup(request-group()); if (!group) { return grpc::Status(grpc::StatusCode::NOT_FOUND, Group not found); } // 在组里面再得到对应的值 auto value group-Get(request-key()); if (!value) { return grpc::Status(grpc::StatusCode::NOT_FOUND, Key not found); } response-set_value(value-ToString()); return grpc::Status::OK; } auto CacheServer::Set(grpc::ServerContext* context, const pb::SetRequest* request, pb::SetResponse* response) - grpc::Status { auto group GetCacheGroup(request-group()); if (!group) { return grpc::Status(grpc::StatusCode::NOT_FOUND, Group not found); } bool is_set group-Set(request-key(), request-value()); response-set_success(is_set); return grpc::Status::OK; } auto Delete(grpc::ServerContext* context, const pb::GetRequest* request, pb::DeleteResponse* response) - grpc::Status { auto group GetCacheGroup(request-group()); if (!group) { return grpc::Status(grpc::StatusCode::NOT_FOUND, Group not found); } bool is_delete group-Delete(request-key()); response-set_success(is_delete); return grpc::Status::OK; } auto Invalidate(grpc::ServerContext* context, const pb::GetRequest* request, pb::InvalidateResponse* response) - grpc::Status { auto group GetCacheGroup(request-group()); if (!group) { return grpc::Status(grpc::StatusCode::NOT_FOUND, Group not found); } // RPC 调用总是来自其他的结点调用InvalidateFromPeer避免循环传播 // 删除这个组里面的key之后就无法找到这个key bool is_invalidate group-InvalidateFromPeer(request-key()); response-set_success(is_invalidate); return grpc::Status::OK; }缓存服务器的启动第一步是先创建里面的sRPCServer这个创建的方式是先构建一个builder对象把所有的配置信息交给builder最后调用grpc_server_ builder.BuildAndStart();完成创建。builder.RegisterService(this);其中这一步是关键我们在cacheServer其实是继承了pb::Cache::Server但是我们重写了getdeleteset等函数所以我们这里要告诉gRPCServer之后再调用对象是我新继承重写的而不是原来的这里传入的this也正是CacheServergrpc_server_-Wait();的作用是阻塞当前线程等待 gRPC 服务器关闭调用Wait()后当前线程停在这一行不继续往下执行。服务器仍然可以接收和处理请求这些工作由 gRPC 内部线程完成。其他线程调用grpc_server_-Shutdown()服务器关闭后Wait()才会返回void CacheServer::Start() { try { // 配置gRPC服务器的选项,把配置信息赋值给builder然后通过builder构建gRPC_server grpc::ServerBuilder builder; builder.SetMaxReceiveMessageSize(opts_.max_msg_size); builder.SetMaxSendMessageSize(opts_.max_msg_size); // 配置TLS或非安全连接 if (opts_.tls) { auto creds LoadTLSCredentials(opts_.cert_file, opts_.key_file); if (!creds) { throw std::runtime_error(Failed to load TLS credentials); } // 绑定监听端口 builder.AddListeningPort(addr_, creds); } else { // grpc::InsecureServerCredentials() 是一个函数返回用于非 TLS 连接的服务器凭据 builder.AddListeningPort(addr_, grpc::InsecureServerCredentials()); } // 启用默认健康检查服务 // 健康检查服务就是让其他程序询问这个服务器现在能正常提供服务吗 grpc::EnableDefaultHealthCheckService(true); builder.SetOption(grpc::MakeChannelArgumentOption(GRPC_ARG_KEEPALIVE_TIME_MS, 30000)); builder.SetOption(grpc::MakeChannelArgumentOption(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 5000)); // 注册服务 // 把当前对象注册为 RPC 请求的处理者让 gRPC 知道收到请求后该调用谁的方法 // 因为我们继承了CacheServer但是重写了里面的很多方法所以需要重新注册我的服务器实现 // this传入的是我们实现的CacheServer builder.RegisterService(this); // 构建并启动服务器 grpc_server_ builder.BuildAndStart(); if (!grpc_server_) { throw std::runtime_error(Failed to build and start gRPC server); } // 从已经创建的 gRPC 服务器中获取健康检查服务的指针 auto health_service grpc_server_-GetHealthCheckService(); if (health_service) { health_service-SetServingStatus(svc_name_, true); } is_stop_ false; std::cout gRPC Server start success at addr_ std::endl; grpc_server_-Wait(); } catch (const std::exception e) { printf(Failed to start gRPC Server: %s\n, e.what()); throw; } }一个CacheServer里面两个关键的部分一个是gRPCServer一个是etcdRegister这个server并不是一个完整的结点而是结点的一部分不包含缓存数据但是可以通过get这些操作来操作缓存数据。总结本篇文章到这里就结束了希望可以帮助大家理解~~~