简介

etcd 是一个分布式键值对存储，设计用来可靠而快速的保存关键数据并提供访问。通过分布式锁，leader选举和写屏障(write barriers)来实现可靠的分布式协作。etcd集群是为高可用，持久性数据存储和检索而准备。

特性

通过官网给出的特性，我们可以看到：

简单的接口: etcd使用标准HTTP接口读写
键值对存储: etcd的核心业务是键值对存储，实际业务中常常用于服务注册和发现
监控变化: etcd能监控特定的key或者目录，并触发回调函数（对于监控服务十分有用）

安装

在宿主机上安装服务

sudo apt update -y
# 安装
sudo apt-get install etcd 
# 启动
sudo systemctl start etcd
# 配置开机自启
sudo systemctl enable etcd

在Docker中运行etcd容器

# 从官方镜像仓库拉取
git pull gcr.io/etcd-development/etcd:v3.5.19
# 从国内镜像源拉取
git pull swr.cn-north-4.myhuaweicloud.com/ddn-k8s/gcr.io/etcd-development/etcd:v3.5.19

编程

我们直入主题，看看怎么编写C++客户端用于调用etcd服务器的服务,实现对服务注册发现功能的封装

由于etcd 采用golang 编写，v3版本通信采用grpc API，即(HTTP2+protobuf),官方只维护了go语言版本的client库, 因此需要找到C/C++ 非官方的client 开发库,这里我们使用etcd-cpp-apiv3

etcd-cpp-apiv3

etcd-cpp-apiv3是一个etcd的C++版本客户端API。它依赖于mipsasm, boost, protobuf, gRPC, cpprestsdk等库。

所以我们先安装依赖

sudo apt update -y
sudo apt install -y libboost-all-dev libssl-dev 
sudo apt install -y libprotobuf-dev protobuf-compiler-grpc 
sudo apt install -y libgrpc-dev libgrpc++-dev  
sudo apt install -y libcpprest-dev

然后再使用源码编译安装

git clone https://github.com/etcd-cpp-apiv3/etcd-cpp-apiv3.git
cd ./etcd-cpp-apiv3
mkdir build && cd ./build
cmake .. -DCMAKE_INSTALL_PREFIX=/usr
make -j$(nproc) && sudo make install

租约系统

这里特别介绍一下etcd的租约系统:

租约(Lease)是etcd中的一种计时机制
每个租约都有TTL(Time To Live)时间
租约到期后，关联的key会自动删除
可以续约(KeeAlive)来延长租约时间

也就是说，在微服务系统中，将服务注册到etcd中后，如果某一个自服务挂了，不再能续约了，那么它的注册在到期后将自动被删除。

常用类和接口介绍

etcd::Client

连接服务端所使用的异步客户端类（同步的叫SyncClient）

常用的构造函数是这个，给出连接服务端的url即可

Client(std::string const& etcd_url,
        std::string const& load_balancer = "round_robin");

//如果需要验证身份就得用这个
Client(std::string const& etcd_url, std::string const& username,
        std::string const& password, int const auth_token_ttl = 300,
        std::string const& load_balancer = "round_robin");

etcd::Client::put:设置键值对,若存在，则覆写；若不存在，则创建

pplx::task<Response> put(std::string const& key, std::string const& value);

pplx::task<Response> put(std::string const& key, std::string const& value,
                         const int64_t leaseId);

etcd::Client::put:设置新键值对,若存在，则失败；若不存在，则创建

1 2	pplx::task<Response> add(std::string const& key, std::string const& value, const int64_t leaseId = 0);

etcd::Client::rm:删除指定的key

1	pplx::task<Response> rm(std::string const& key);

etcd::Client::watch: 监控指定的key或目录的值是否变化,若要监控目标目录的子目录，则要将recursive置为true
1
pplx::task<Response> watch(std::string const& key, bool recursive = false);
etcd::Client::leasekeepalive: 创建租约保持器，ttl为租约/过期时间
1
pplx::task<std::shared_ptr<KeepAlive>> leasekeepalive(int ttl);

etcd::KeepAlive

这是一个租约保持器，专门用于给键值对保活（类似于心跳包）。

etcd::KeepAlive::Lease: 获取lease_id

1	int64_t Lease() const { return lease_id; }

``etcd::KeepAlive::Cancel`: 取消保活功能（之后过段时间就会自动过期）
1
void Cancel();

编写Registry类

我们将服务注册功能封装进Registry类中，并提供注册服务接口

namespace btyGoose
{
//服务注册类
class Registry{
    
    Registry(const std::string& etcd_host)
    :_client(std::make_shared<etcd::Client>(etcd_host))
    ,_keep_alive(_client->leasekeepalive(3).get())
    ,_lease_id(_keep_alive->Lease())        //开始续租
    {}

    ~Registry() { _keep_alive->Cancel(); }//析构函数自动停止续租

    bool registerService(const std::string&key,const std::string& value)
    {
        auto resp = _client->put(key,value,_lease_id).get();
        if(resp.is_ok() == false)//如果失败了
        {
            SUP_LOG_ERROR("注册服务失败: {}",resp.error_message());
            return false;
        }
        //成功注册
        return true;
    }

    uint64_t leaseId()
    {
        return _lease_id;
    }
private:
    std::shared_ptr<etcd::Client> _client;
    std::shared_ptr<etcd::KeepAlive> _keep_alive;
    uint64_t _lease_id;
};
}

编写Discovery类

我们将服务发现功能封装进Discovery类中，同时也将监控功能封装在里面,以便及时响应服务的上下线的变化

它主要执行如下两大功能:

初始化时，获取已经注册的服务
初始化后，持续监控服务的变化,并触发对应的回调函数

class Discovery
{
public:
    using NotifyCallback= std::function<void(std::string,std::string)> ;

    Discovery(
        const std::string& etcd_host,
        const std::string& basedir,
        const NotifyCallback& put_cb,
        const NotifyCallback& del_cb 
    )
    :_client(std::make_shared<etcd::Client>(etcd_host))
    ,_put_callback(put_cb)
    ,_del_callback(del_cb)
    {
        //先获取已有的服务
        auto resp = _client->ls(basedir).get();
        if(resp.is_ok() == false)
        {
            SUP_LOG_ERROR("Discovery获取服务信息失败: {}",resp.error_message());
        }
        int sz = resp.keys().size();
        SUP_LOG_INFO("Discovery初始化发现已有服务数： {}",sz);
        for(int i = 0;i<sz;++i)
        {
            if(_put_callback)//如果设置了回调函数
            {
                _put_callback(resp.key(i),resp.value(i).as_string());
                SUP_LOG_DEBUG("Discovery初始化时，发现服务： {}->{}",resp.key(i),resp.value(i).as_string());
            }
        }
        //然后监控后续的变化,注意recursive要为true
        _watcher = std::make_shared<etcd::Watcher>(*(_client.get()),basedir,
            std::bind(&Discovery::callback,this,std::placeholders::_1),true);
    }

    ~Discovery()
    {
        _watcher->Cancel();//取消监控
    }

    void callback(const etcd::Response &resp)
    {
        if(resp.is_ok() == false)
        {
            SUP_LOG_ERROR("收到一个错误事件的通知： {}",resp.error_message());
            return;
        }

        for(auto const&ev:resp.events())
        {
            if(ev.event_type() == etcd::Event::EventType::PUT)
            {
                //如果设置了回调函数
                if(_put_callback)
                {
                    _put_callback(ev.kv().key(),ev.kv().as_string());
                    SUP_LOG_DEBUG("新增服务: {}->{}",ev.kv().key(),ev.kv().as_string());
                }
            }
            else if (ev.event_type() == etcd::Event::EventType::DELETE_)
            {
                //如果设置了回调函数
                if(_del_callback)
                {
                    _del_callback(ev.kv().key(),ev.kv().as_string());
                    SUP_LOG_DEBUG("下线服务： {}->{}",ev.kv().key(),ev.kv().as_string());
                }
            }
        }
    }

private:
    NotifyCallback _put_callback;
    NotifyCallback _del_callback;
    std::shared_ptr<etcd::Client> _client;
    std::shared_ptr<etcd::Watcher> _watcher;
};

测试代码

我们来写一个main.cpp源文件来测试我们封装好的代码,其中有如下要点:

注意在注册不同的服务时，键要及时改变，否则会前后覆盖
Discovery的生命周期必须大于Registry的生命周期+租约过期时间
需要自行提供两个回调函数

代码实现如下:

#include "etcd.hpp"
#include <unordered_map>
//服务端的url
const string& etcd_host = "http://127.0.0.1:2379";
std::unordered_map<std::string,std::string> service_table;

void put_cb(std::string key,std::string value)
{
    service_table[key] = value;
    SUP_LOG_INFO("已存入服务列表: {}->{}",key,value);
}
void del_cb(std::string key,std::string value)
{
    service_table.erase(key);
    SUP_LOG_INFO("已下线服务: {}",key);
}

void testRegister()
{
    btyGoose::Registry* rs = new btyGoose::Registry(etcd_host);
    rs->registerService("example_service1-1","http://example.ex:8888");
    rs->registerService("example_service1-2","http://example.ex:9999");
    rs->registerService("example_service2-1","http://example2.ex:8888");

    btyGoose::Discovery dc(etcd_host,"",put_cb,del_cb);
    SUP_LOG_INFO("Discovery已创建");
    sleep(1);

    rs->registerService("example_service3-3","http://testDiscovery.ex:8888");

    for(auto [x,y]:service_table)
    {
        SUP_LOG_DEBUG("Register析构前含有服务: {}->{}",std::string(x),std::string(y));
    }
    delete rs;

    SUP_LOG_INFO("接下来等待10秒...");
    
    sleep(10);
    int size = service_table.size();
    SUP_LOG_INFO("睡醒了！来看看还剩多少服务记录: {}",size);
}

int main()
{
    std::cerr<<"test\n";
    btyGoose::init_logger();
    SUP_LOG_INFO("开始测试");
    testRegister();


    return 0;
}

对应的makefile如下

mycmd: main.o logger.o
	g++ -o $@ *.o -std=c++17 -lpthread -letcd-cpp-api -lspdlog -lssl -lcrypto -lgrpc++ -lprotobuf -lfmt -lcpprest

main.o:main.cpp
	g++ -c main.cpp -o main.o

logger.o:logger.cpp
	g++ -c logger.cpp -o logger.o

.PHONY:clean
clean:
	rm -f mycmd *.o

测试结果

可以看到它们成功地完成了服务的注册，发现和监控功能

supdriver@ALi-cloud-Linux-2-2G:~/codes/Linux_Code/etcdTest$ ./mycmd 
test
[default-logger][14:22:20][2256873][info    ][main.cpp:48] 开始测试
[default-logger][14:22:20][2256873][debug   ][etcd.hpp:32] 成功注册服务: example_service1-1-http://example.ex:8888
[default-logger][14:22:20][2256873][debug   ][etcd.hpp:32] 成功注册服务: example_service1-2-http://example.ex:9999
[default-logger][14:22:20][2256873][debug   ][etcd.hpp:32] 成功注册服务: example_service2-1-http://example2.ex:8888
[default-logger][14:22:22][2256873][info    ][etcd.hpp:68] Discovery初始化发现已有服务数： 3
[default-logger][14:22:22][2256873][info    ][main.cpp:10] 已存入服务列表: example_service1-1->http://example.ex:8888
[default-logger][14:22:22][2256873][debug   ][etcd.hpp:74] Discovery初始化时，发现服务： example_service1-1->http://example.ex:8888
[default-logger][14:22:22][2256873][info    ][main.cpp:10] 已存入服务列表: example_service1-2->http://example.ex:9999
[default-logger][14:22:22][2256873][debug   ][etcd.hpp:74] Discovery初始化时，发现服务： example_service1-2->http://example.ex:9999
[default-logger][14:22:22][2256873][info    ][main.cpp:10] 已存入服务列表: example_service2-1->http://example2.ex:8888
[default-logger][14:22:22][2256873][debug   ][etcd.hpp:74] Discovery初始化时，发现服务： example_service2-1->http://example2.ex:8888
[default-logger][14:22:22][2256873][info    ][main.cpp:26] Discovery已创建
[default-logger][14:22:23][2256934][info    ][main.cpp:10] 已存入服务列表: example_service3-3->http://testDiscovery.ex:8888
[default-logger][14:22:23][2256934][debug   ][etcd.hpp:103] 新增服务: example_service3-3->http://testDiscovery.ex:8888
[default-logger][14:22:23][2256873][debug   ][etcd.hpp:32] 成功注册服务: example_service3-3-http://testDiscovery.ex:8888
[default-logger][14:22:23][2256873][debug   ][main.cpp:33] Register析构前含有服务: example_service3-3->http://testDiscovery.ex:8888
[default-logger][14:22:23][2256873][debug   ][main.cpp:33] Register析构前含有服务: example_service2-1->http://example2.ex:8888
[default-logger][14:22:23][2256873][debug   ][main.cpp:33] Register析构前含有服务: example_service1-2->http://example.ex:9999
[default-logger][14:22:23][2256873][debug   ][main.cpp:33] Register析构前含有服务: example_service1-1->http://example.ex:8888
[default-logger][14:22:23][2256873][info    ][main.cpp:37] 接下来等待10秒...
[default-logger][14:22:25][2256934][info    ][main.cpp:15] 已下线服务: example_service1-1
[default-logger][14:22:25][2256934][debug   ][etcd.hpp:112] 下线服务： example_service1-1
[default-logger][14:22:25][2256934][info    ][main.cpp:15] 已下线服务: example_service1-2
[default-logger][14:22:25][2256934][debug   ][etcd.hpp:112] 下线服务： example_service1-2
[default-logger][14:22:25][2256934][info    ][main.cpp:15] 已下线服务: example_service2-1
[default-logger][14:22:25][2256934][debug   ][etcd.hpp:112] 下线服务： example_service2-1
[default-logger][14:22:25][2256934][info    ][main.cpp:15] 已下线服务: example_service3-3
[default-logger][14:22:25][2256934][debug   ][etcd.hpp:112] 下线服务： example_service3-3
[default-logger][14:22:33][2256873][info    ][main.cpp:41] 睡醒了！来看看还剩多少服务记录: 0
[warn] watcher does't exit normally