项目组件框架介绍[etcd]

news2024/12/13 21:23:06

文章目录

前言
etcd安装
- Ubuntu 上通过包管理器安装
- 通过源码安装
- 配置
客户端开发包
- 开发包的安装
- 接口介绍
- - 添加一个键值对
  - 获取一个键值对
  - 租约保活机制
  - 监听
封装服务注册与发现
- 服务注册
- 服务发现

前言

Etcd 是一个 golang 编写的分布式、高可用的一致性键值存储系统，用于配置共享和服务发现等

etcd安装

Ubuntu 上通过包管理器安装

# 直接安装
sudo apt-get install etcd
# 启动
sudo systemctl start etcd
# 设置开机自启
sudo systemctl enable etcd

验证安装

$ etcd --version
etcd Version: 3.2.26
Git SHA: Not provided (use ./build instead of go build)
Go Version: go1.13.8
Go OS/Arch: linux/amd64

可见已经安装成功

通过源码安装

etcd是使用Go语言编写的, 根据官网描述的教程, 需要Go1.2+的版本

# 克隆仓库
$ git clone -b v3.5.16 https://github.com/etcd-io/etcd.git
$ cd etcd
# 运行构建脚本：
$ ./build.sh
# 二进制文件位于 directory/bin
# 设置环境变量, 假如你的二进制文件位于/bin
$ export PATH="$PATH:`pwd`/bin"
# 验证安装
$ etcd --version

配置

默认 etcd 的集群节点通信端口为 2380, 客户端访问端口为 2379, 如果需要更改可以修改/etc/default/etcd文件
修改配置

客户端开发包

开发包的安装

官方只维护了 go 语言版本的 client 库, 因此需要使用 C/C++ 非官方的 client 开发库, etcd-cpp-apiv3是一个不错的选择, etcd-cpp-apiv3 是一个 etcd 的 C++版本客户端 API。它依赖于 mipsasm, boost, protobuf, gRPC, cpprestsdk 等库
所有首先需要安装依赖

# 建议先看看是否已经用其他方式安装这些, 否则出现版本冲突问题很难受
# 比如up就在项目中途把protobuf版本更新了, 就导致所有依赖protobuf编译安装的第三方库都用不了
# 比如这个, brpc等等都用不了,最后重新编译安装了它们才解决, 还出现了莫名其妙的库找不到的问题
sudo apt-get install libboost-all-dev libssl-dev
sudo apt-get install libprotobuf-dev protobuf-compiler-grpc
sudo apt-get install libgrpc-dev libgrpc++-dev 
sudo apt-get install 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进行服务注册和发现, 我们重点介绍这些接口, 其他介绍可参考官方介绍

添加一个键值对

#include<etcd/Client.hpp>

//构造一个etcd::Client 对象, 用etcd服务端的url构造
etcd::Client etcd("http://127.0.0.1:2379");
//使用put方法添加一个键值对, etcd.put() 方法是非阻塞的，返回一个异步任务pplx::task
pplx::task<etcd::Response> response_task = etcd.put("/test/1", "111");
// 等待异步任务完成并获取结果
try 
{
    etcd::Response response = response_task.get();
    if (response.is_ok()) 
    {
        std::cout << "Put: " << response.key() << " : " << response.value() << std::endl;
    } 
    else 
    {
        std::cerr << "Error: " << response.error_message() << std::endl;
    }
} 
catch (const std::exception &e) 
{
    std::cerr << e.what() << std::endl;
}

获取一个键值对

etcd::Client etcd("http://127.0.0.1:2379");
//使用get方法获取一个键值对, 这个也是非阻塞的, 返回一个异步任务
pplx::task<etcd::Response> response_task = etcd.get("/test/1");
try
{
    etcd::Response response = response_task.get();
    if (response.is_ok())
    {
        std::cout << "获取成功" << response.value().as_string();
    }
    else
    {
        std::cout << "获取失败: " << response.error_message();
    }
}
catch (std::exception const & ex)
{
    std::cerr << ex.what() <<std::endl;
}

租约保活机制

在项目中我们需要监控一个服务的在线情况, 如果异常退出, 有时候就不能很好的检测到, etcd提供了租约保活机制, 服务如果没有续约, 就可以认为该服务已下线

    etcd::Client etcd("http://127.0.0.1:2379");
    //获取一个三秒的租约保活对象 
    std::shared_ptr<etcd::KeepAlive> keep(etcd.leasekeepalive(3).get());
    //获取租约ID
    int64 lease_id=keep->Lease();
    etcd.put("/test/2", "222",lease_id);

监听

项目中需要对服务的上线与下线进行监控, 再结合其他组件, 比如bprc, 就可通过这个服务获取相对应的通信对象

    //构造一个监控对象, 第二个参数可以是一个目录, 比如/test, 那么就会监控/test下比如/test/1 /test/2, 第三个是一个回调函数, 当被监控的对象发送改变时就会触发回调
    etcd::Watcher watcher("http://127.0.0.1:2379", "/test", CallBack);
    //etcd::Response是一个容器, 可以通过遍历的方式获取, 比如
    void CallBack(const etcd::Response& re)
    {
        if(!re.is_ok())
        {
            LOG_ROOT_ERROR<<re.error_message();
            return;
        }
        for(const auto& ev: re.events())
        {
            //通过etcd::EVent::EventType 可得知具体情况
            if(ev.event_type()==etcd::Event::EventType::PUT)
            {
                LOG_ROOT_INFO<<"新增服务: "<<ev.kv().key()<<" : "<<ev.kv().as_string();
            }
            if(ev.event_type()==etcd::Event::EventType::DELETE_)
            {
                LOG_ROOT_INFO<<"服务下线: "<<ev.prev_kv().key()<<" : "<<ev.prev_kv().as_string();
            }
        }
    }

封装服务注册与发现

有了上面了了解, 我们就能完成一个简单的服务注册与发现模块

服务注册

#include<etcd/Client.hpp>
#include<etcd/KeepAlive.hpp>
#include<etcd/Response.hpp>
#include<etcd/Value.hpp>
#include<etcd/Watcher.hpp>
#include"log.hpp"
namespace MindbniM
{
    class Registry
    {
    public:
        using ptr=std::shared_ptr<Registry>;
        //这里默认使用3秒保活了, 也可以手动传入
        Registry(const std::string& host):_client(std::make_unique<etcd::Client>(host)),_ka(_client->leasekeepalive(3).get()),_lease_id(_ka->Lease())
        {}
        bool registry(const std::string& key,const std::string& value)
        {
            etcd::Response re=_client->put(key,value,_lease_id).get();
            if(!re.is_ok())
            {
                LOG_ROOT_ERROR<<"服务注册失败"<<":"<<re.error_message();
                return false;
            }
            LOG_ROOT_DEBUG<<"服务注册:"<<key<<" : "<<value;
            return true;
        }
    private:
        std::unique_ptr<etcd::Client> _client;
        std::shared_ptr<etcd::KeepAlive> _ka;
        int64_t _lease_id;
    };
}

服务发现

为了和其他组件相互结合, 我们设置两个回调, 分别对应服务上线与下线的操作

#include<etcd/Client.hpp>
#include<etcd/Response.hpp>
#include<etcd/Value.hpp>
#include<etcd/Watcher.hpp>
#include"log.hpp"
namespace MindbniM
{
    class Discovery
    {
    public:
        using ptr=std::shared_ptr<Discovery>;
        //对应key和value
        using CallBack=std::function<void(const std::string&,const std::string&)>;
        Discovery(const std::string& host,const CallBack& put=nullptr,const CallBack& del=nullptr):_client(std::make_unique<etcd::Client>(host)),_put(put),_del(del)
        {}
        bool discover(const std::string& dir)
        {
            etcd::Response re=_client->ls(dir).get();
            if(!re.is_ok())
            {
                LOG_ROOT_ERROR<<"服务发现错误"<<re.error_message();
                return false;
            }
            int n=re.keys().size();
            if(_put!=nullptr)
            {
                for(int i=0;i<n;i++)
                {
                    _put(re.key(i),re.value(i).as_string());
                }
            }
            //开始监听
            _watch=std::make_unique<etcd::Watcher>(*_client,dir,std::bind(&Discovery::_CallBack,this,std::placeholders::_1),true);
            return true;
        }
        bool wait()
        {
            return _watch->Wait();
        }
    private:
        void _CallBack(const etcd::Response& re)
        {
            if(!re.is_ok())
            {
                LOG_ROOT_ERROR<<re.error_message();
                return;
            }
            for(const auto& ev: re.events())
            {
                if(ev.event_type()==etcd::Event::EventType::PUT)
                {
                    if(_put) _put(ev.kv().key(),ev.kv().as_string());
                    LOG_ROOT_INFO<<"新增服务: "<<ev.kv().key()<<" : "<<ev.kv().as_string();
                }
                if(ev.event_type()==etcd::Event::EventType::DELETE_)
                {
                    if(_del) _del(ev.prev_kv().key(),ev.prev_ky().as_string());
                    LOG_ROOT_INFO<<"服务下线: "<<ev.prev_kv().key()<<" : "<<ev.prev_kv().as_string();
                }
            }
        }
        CallBack _put;
        CallBack _del;
        std::unique_ptr<etcd::Client> _client;
        std::unique_ptr<etcd::Watcher> _watch;
    };
}