基于微服务的即时通讯系统 -- etcd实现服务注册与发现

前言

我们首先通过一个架构图来理解即时通讯系统：

即时通讯系统是高并发高可用的，倘若不加注册中心，网关需要轮询每个子服务，子服务需要主动通知网关自己准备好了，不说效率，只是注册发现都会存在海量网络传输，引入注册中心将发现和注册解耦，极大提高效率。

etcd实现注册中心

etcd是一个存储键值对的服务器（可以理解为数据库），该项目中引入了etcd实现注册中心，根本原因是其中的高级特性与机制能够解决多个子服务的数据一致性，换句话说，它可以监听一个服务的变化从而通知子服务，这在多个子服务的场景下十分重要，否则注册中心只有一个服务备份，谁更改了它，其他子服务都无从得知。

（1）主机与etcd

从主机与etcd的角度看，主机 a 向 etcd 中注册一个 (服务名，提供服务的主机地址）键值对，

主机 b 从etcd中申请到可用的服务。

（2）etcd - CP架构

那etcd底层如何实现最重要的数据一致性呢，注册流程保持一致，不过在注册接口 put 中我们注意到一个 lease_id ，它是租约ID，什么是租约？简单来说就是键值对数据能存储在 etcd 中的时间，通过创建一个租约对象就能让键值对数据在租约过后自动删除。

如果想续租呢？通常使用一个Keepalive（长连接对象）为租约续租，其实就是为 etcd 客户端和外部注册服务的主机建立一个长连接，一旦外部服务下线了，长连接自然也断开，从而通知 etcd 服务端删除该服务。

监听客户端如何知道这样的变化呢？它首先从 etcd 服务端拉取在 base_service 目录下的服务，之后交给 Watcher 对象监听，Watcher 对象监听到服务变化后，通过回调函数反馈给监听客户端。

注册服务客户端代码示例：

// 注册服务的客户端类 class Registery { public: // 创建一个智能指针便于外部访问 using ptr = std::shared_ptr<Registery>; Registery(const std::string& registery_host) :_registery_host(registery_host), _put_client(std::make_shared<etcd::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_ptr<etcd::Client> _put_client; // 注册客户端 std::shared_ptr<etcd::KeepAlive> _keepalive; // 长连接租约保活对象 uint64_t _lease_id; // 租约id };

监听服务客户端代码示例：

// 监听(发现)服务的客户端类 // 提供两个put和get回调函数，以此让用户做出响应 using callback = std::function<void(etcd::Response)>; class Discovery { public: // 创建一个智能指针便于外部访问 using ptr = std::shared_ptr<Discovery>; using NotifyCallback = std::function<void(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_shared<etcd::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_shared<etcd::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_ptr<etcd::Client> _get_client; // 监听客户端 std::shared_ptr<etcd::Watcher> _watcher; // 监听对象 };