前言我们首先通过一个架构图来理解即时通讯系统即时通讯系统是高并发高可用的倘若不加注册中心网关需要轮询每个子服务子服务需要主动通知网关自己准备好了不说效率只是注册发现都会存在海量网络传输引入注册中心将发现和注册解耦极大提高效率。etcd实现注册中心etcd是一个存储键值对的服务器可以理解为数据库该项目中引入了etcd实现注册中心根本原因是其中的高级特性与机制能够解决多个子服务的数据一致性换句话说它可以监听一个服务的变化从而通知子服务这在多个子服务的场景下十分重要否则注册中心只有一个服务备份谁更改了它其他子服务都无从得知。1主机与etcd从主机与etcd的角度看主机 a 向 etcd 中注册一个 (服务名提供服务的主机地址键值对主机 b 从etcd中申请到可用的服务。2etcd - CP架构那etcd底层如何实现最重要的数据一致性呢注册流程保持一致不过在注册接口 put 中我们注意到一个 lease_id 它是租约ID什么是租约简单来说就是键值对数据能存储在 etcd 中的时间通过创建一个租约对象就能让键值对数据在租约过后自动删除。如果想续租呢通常使用一个Keepalive长连接对象为租约续租其实就是为 etcd 客户端和外部注册服务的主机建立一个长连接一旦外部服务下线了长连接自然也断开从而通知 etcd 服务端删除该服务。监听客户端如何知道这样的变化呢它首先从 etcd 服务端拉取在 base_service 目录下的服务之后交给 Watcher 对象监听Watcher 对象监听到服务变化后通过回调函数反馈给监听客户端。注册服务客户端代码示例// 注册服务的客户端类 class Registery { public: // 创建一个智能指针便于外部访问 using ptr std::shared_ptrRegistery; Registery(const std::string registery_host) :_registery_host(registery_host), _put_client(std::make_sharedetcd::Client(registery_host)), _keepalive(_put_client-leasekeepalive(3).get()), _lease_id(_keepalive-Lease()) { // 先创建客户端 // 再创建长连接保活对象 // 最后得到租约id } bool registery(const std::string key, const std::string val) { auto resp _put_client-put(key, val, _lease_id).get(); if(resp.is_ok() false) { LOG_ERROR({} 注册的 {}服务失败\n, val, key); return false; } LOG_DEBUG({} 注册的 {}服务成功\n, val, key) return true; } ~Registery() { LOG_DEBUG(主机{}已下线\n, _registery_host); _keepalive-Cancel();} private: std::string _registery_host; // 主机名 std::shared_ptretcd::Client _put_client; // 注册客户端 std::shared_ptretcd::KeepAlive _keepalive; // 长连接租约保活对象 uint64_t _lease_id; // 租约id };监听服务客户端代码示例// 监听(发现)服务的客户端类 // 提供两个put和get回调函数以此让用户做出响应 using callback std::functionvoid(etcd::Response); class Discovery { public: // 创建一个智能指针便于外部访问 using ptr std::shared_ptrDiscovery; using NotifyCallback std::functionvoid(std::string, std::string); Discovery(const std::string registery_host, const std::string base_dir, const NotifyCallback put_cb, const NotifyCallback del_cb) :_registery_host(registery_host), _get_client(std::make_sharedetcd::Client(_registery_host)), _base_dir(base_dir), _put_cb(put_cb), _del_cb(del_cb) { // 2、拉取所有注册过的服务 auto resp _get_client-ls(_base_dir).get(); if(resp.is_ok()) { // 遍历所有注册的服务 for(int i 0; i resp.keys().size(); i) { std::cout resp.key(i) 可以提供 resp.value(i) 服务 std::endl; // 发现服务并触发回调 if(_put_cb) _put_cb(resp.key(i), resp.value(i).as_string()); } } else { LOG_ERROR(获取键值对数据失败-{}\n, resp.error_message()); } // 3、创建监听对象 _watcher std::make_sharedetcd::Watcher(*_get_client.get(), _base_dir, std::bind(Discovery::WatcherCallback, this, std::placeholders::_1)); _watcher-Wait(); } private: void WatcherCallback(const etcd::Response resp) { if (resp.is_ok() false) { std::cout 获取到一个错误的事件通知 - resp.error_message() std::endl; return; } for(const auto ev : resp.events()) { if(ev.event_type() etcd::Event::EventType::PUT) { std::cout 新增服务 ev.kv().key() - ev.kv().as_string() std::endl; if(_put_cb) _put_cb(ev.kv().key(), ev.kv().as_string()); } else if(ev.event_type() etcd::Event::EventType::DELETE_) { std::cout 删除服务 ev.prev_kv().key() - ev.prev_kv().as_string() std::endl; if(_del_cb) _del_cb(ev.prev_kv().key(), ev.prev_kv().as_string()); } } } private: NotifyCallback _put_cb; // 注册回调 NotifyCallback _del_cb; // 发现回调 std::string _registery_host; // 主机名 std::string _base_dir; // 监听目录 std::shared_ptretcd::Client _get_client; // 监听客户端 std::shared_ptretcd::Watcher _watcher; // 监听对象 };