文章目录
- 写在前面
- 基本使用
- 封装客户端
- 封装服务端
- Zookeeper
写在前面
最近学了一下gRPC进行远程调用的原理，所以把这个项目改造成了微服务分布式的架构，今天也是基本实现好了，代码已提交
这里补充一下文档吧，也算记录一下整个过程
基本使用
gRPC首先在安装上就非常繁琐，网络的教程也比较多，但要注意安装的版本兼容性问题，尤其是对应的Protubuf和gRPC的版本，同时要注意，在进行编译的时候要使用cmake进行编译，我最开始使用的是传统的Makefile，因为项目最开始用的就是这种，所以就直接使用了，而在进行编译链接的时候总是报错：
最后去查阅了官方文档，也就是gRPC的维护者，文档提示最好使用cmake进行编译：
https://github.com/grpc/grpc/tree/master/src/cpp
用了cmake就不会报链接的错误了，总体来说，gRPC安装确实繁琐，需要细心一点
使用的命令也比较简单：
protoc --grpc_out=. --plugin=protoc-gen-grpc=`which grpc_cpp_plugin` message.proto
protoc --cpp_out=. message.proto
一个是生成grpc文件的，一个是生成proto文件的
下面我以一个服务为演示吧，其他的服务基本差不多，我就不再演示了
封装客户端
首先，对于客户端进行封装：
#pragma once
#include "const.h"
#include "Singleton.h"
#include "ConfigMgr.h"
#include <condition_variable>
#include <grpcpp/grpcpp.h>
#include <queue>
#include "message.grpc.pb.h"
#include "message.pb.h"
using grpc::Channel;
using grpc::Status;
using grpc::ClientContext;
using message::LoginRsp;
using message::LoginReq;
using message::StatusService;
/**
* @brief 管理 gRPC 客户端的连接池，用于与 StatusService 通信
*
*/
class StatusConPool
{
public:
StatusConPool(size_t poolSize, std::string host, std::string port)
: poolSize_(poolSize), host_(host), port_(port), b_stop_(false)
{
for (size_t i = 0; i < poolSize_; ++i)
{
std::shared_ptr<Channel> channel = grpc::CreateChannel(host + ":" + port,
grpc::InsecureChannelCredentials());
connections_.push(StatusService::NewStub(channel));
}
}
~StatusConPool()
{
std::lock_guard<std::mutex> lock(mutex_);
Close();
while (!connections_.empty())
{
connections_.pop();
}
}
std::unique_ptr<StatusService::Stub> getConnection()
{
std::unique_lock<std::mutex> lock(mutex_);
cond_.wait(lock, [this] {
if (b_stop_)
{
return true;
}
return !connections_.empty();
});
//如果停止则直接返回空指针
if (b_stop_)
{
return nullptr;
}
auto context = std::move(connections_.front());
connections_.pop();
return context;
}
void returnConnection(std::unique_ptr<StatusService::Stub> context)
{
std::lock_guard<std::mutex> lock(mutex_);
if (b_stop_)
{
return;
}
connections_.push(std::move(context));
cond_.notify_one();
}
void Close()
{
b_stop_ = true;
cond_.notify_all();
}
private:
atomic<bool> b_stop_;
size_t poolSize_;
std::string host_;
std::string port_;
std::queue<std::unique_ptr<StatusService::Stub>> connections_;
std::mutex mutex_;
std::condition_variable cond_;
};
/**
* @brief 通过单例模式实现的 gRPC 客户端，用于向 StatusService 发送请求
*
*/
class StatusGrpcClient :public Singleton<StatusGrpcClient>
{
friend class Singleton<StatusGrpcClient>;
public:
~StatusGrpcClient()
{}
LoginRsp Login(string username, string password)
{
ClientContext context;
LoginRsp reply;
LoginReq request;
request.set_username(username);
request.set_password(password);
auto stub = pool_->getConnection();
cout << "准备进行grpc 发送了" << endl;
Status status = stub->Login(&context, request, &reply);
Defer defer([&stub, this]() {
pool_->returnConnection(std::move(stub));
});
if (status.ok())
{
return reply;
}
else
{
reply.set_error(ErrorCodes::RPCFailed);
return reply;
}
}
private:
StatusGrpcClient()
{
auto& gCfgMgr = ConfigMgr::Inst();
std::string host = gCfgMgr["StatusServer"]["Host"];
std::string port = gCfgMgr["StatusServer"]["Port"];
cout << "host:port" << host + ":" + port << endl;
pool_.reset(new StatusConPool(5, host, port));
}
std::unique_ptr<StatusConPool> pool_;
};
在这样进行封装了之后：
由于这里存在对应的接口，此时就能够进行远程调用了，然后对于远程调用回来的结果进行判断即可
封装服务端
gRPC比较优秀的一点就在于，它能够屏蔽网络的传输，使得使用者可以专注的对于业务逻辑进行处理，具体可以看下面这个：
这里proto会生成一个服务类，这个类是一个虚基类，只需要对于这个类进行继承后，实现对应的接口，那么在进行调用的时候就可以去调用我们实际要进行处理的逻辑，就是一个多态的思想：
#pragma once
#include <grpcpp/grpcpp.h>
#include "message.grpc.pb.h"
#include <mutex>
#include "ConfigMgr.h"
#include "MysqlMgr.h"
#include "const.h"
#include "RedisMgr.h"
#include <climits>
#include <nlohmann/json.hpp>
#include <regex>
using grpc::ServerContext;
using grpc::Status;
using message::LoginReq;
using message::LoginRsp;
using message::RegReq;
using message::RegRsp;
using message::StatusService;
using json = nlohmann::json;
class StatusServiceImpl final : public StatusService::Service
{
public:
StatusServiceImpl()
{}
Status Login(ServerContext* context, const LoginReq* request, LoginRsp* reply)
{
cout << "收到了 Login" << endl;
auto username = request->username();
auto password = request->password();
bool success = authenticate(username.c_str(), password.c_str());
cout << "验证成功" << endl;
if(!success)
{
reply->set_error(ErrorCodes::PasswdErr);
cout << "发送成功" << endl;
return Status::OK;
}
reply->set_error(ErrorCodes::Success);
cout << "发送成功" << endl;
return Status::OK;
}
// 验证用户名和密码是否正确
bool authenticate(const char *username, const char *password)
{
cout << "去Redis里面看看" << endl;
if(FindInRedis(username, password))
return true;
cout << "去Mysql里面看看" << endl;
return MysqlMgr::GetInstance()->CheckPwd(username, password);
}
bool FindInRedis(const char* username, const char* password)
{
string result = RedisMgr::GetInstance()->HGet("user:username:password", username);
return result == password;
}
Status Register(ServerContext* context, const RegReq* request, RegRsp* reply)
{
auto username = request->username();
auto password = request->password();
bool success = RegisterInfo(username.c_str(), password.c_str());
if(!success)
{
reply->set_error(ErrorCodes::PasswdErr);
return Status::OK;
}
reply->set_error(ErrorCodes::Success);
return Status::OK;
}
bool validateCredentials(const string& username, const string& password)
{
// 定义用户名的正则表达式
regex usernamePattern("^[a-zA-Z0-9._-]{3,}$");
// 定义密码的正则表达式
regex passwordPattern("^[a-zA-Z0-9._-]{6,}$");
// 使用regex_match进行匹配，注意这里应该是&&操作，因为两个条件都需要满足
if(regex_match(username, usernamePattern) && regex_match(password, passwordPattern))
return true; // 如果都匹配成功，则返回true
else
return false; // 否则返回false
}
// 尝试插入用户信息，成功返回 true，失败返回 false
bool RegisterInfo(const char *username, const char *password)
{
if(!validateCredentials(username, password))
return false;
return MysqlMgr::GetInstance()->RegUser(username, password);
}
};
这样，在外部服务端，就可以进行调用了：
Zookeeper
分布式架构当中存在一个有用的组件，Zookeeper，这个原理是进行一个类似于文件系统的架构，然后可以进行读取其中的值，并且还设置了对应的回调函数，也就是所谓的Watcher，发现有服务到达的时候，就执行对应的回调函数，那么基于这个原理，就可以去动态识别到gRPC的服务
gRPC的服务我也封装好了，其他的就看仓库里面的代码吧
#ifndef _ZOOKEEPER_H_
#define _ZOOKEEPER_H_
#include <zookeeper/zookeeper.h>
#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/noncopyable.hpp>
#include <stdio.h>
#include <string>
#include <vector>
#include <map>
#include <sstream>
#include <typeinfo>
typedef boost::function<void (const std::string &path, const std::string &value)> DataWatchCallback;
typedef boost::function<void (const std::string &path, const std::vector<std::string> &value)> ChildrenWatchCallback;
//
class ZkRet
{
friend class ZooKeeper;
public:
bool ok() const {return ZOK == code_; }
bool nodeExist() const {return ZNODEEXISTS == code_; }
bool nodeNotExist() const {return ZNONODE == code_; }
operator bool() const {return ok(); }
protected:
ZkRet(){code_ = ZOK; }
ZkRet(int c){code_ = c; }
private:
int code_;
};
// class Zookeeper,
// thread safety: single ZooKeeper object should be used in single thread.
class ZooKeeper : public boost::noncopyable
{
public:
ZooKeeper();
~ZooKeeper();
//
ZkRet init(const std::string &connectString);
ZkRet getData(const std::string &path, std::string &value);
ZkRet setData(const std::string &path, const std::string &value);
ZkRet getChildren(const std::string &path, std::vector<std::string> &children);
ZkRet exists(const std::string &path);
ZkRet createNode(const std::string &path, const std::string &value, bool recursive = true);
// ephemeral node is a special node, its has the same lifetime as the session
ZkRet createEphemeralNode(const std::string &path, const std::string &value, bool recursive = true);
// sequence node, the created node's name is not equal to the given path, it is like "path-xx", xx is an auto-increment number
ZkRet createSequenceNode(const std::string &path, const std::string &value, std::string &rpath, bool recursive = true);
ZkRet createSequenceEphemeralNode(const std::string &path, const std::string &value, std::string &rpath, bool recursive = true);
ZkRet watchData(const std::string &path, const DataWatchCallback &wc);
ZkRet watchChildren(const std::string &path, const ChildrenWatchCallback &wc);
//
void setDebugLogLevel(bool open = true);
//
ZkRet setFileLog(const std::string &dir = "./");
ZkRet setConsoleLog();
//
static std::string getParentPath(const std::string &path);
static std::string getNodeName(const std::string &path);
static std::string getParentNodeName(const std::string &path);
private:
// for inner use, you should never call these function
void setConnected(bool connect = true){connected_ = connect; }
bool connected()const{return connected_; }
void restart();
//
// watch class
class Watch
{
public:
Watch(ZooKeeper *zk, const std::string &path);
virtual void getAndSet() const = 0;
const std::string &path() const{return path_; }
ZooKeeper* zk() const {return zk_; }
protected:
ZooKeeper *zk_;
std::string path_;
};
typedef boost::shared_ptr<Watch> WatchPtr;
class DataWatch: public Watch
{
public:
typedef DataWatchCallback CallbackType;
DataWatch(ZooKeeper *zk, const std::string &path, const CallbackType &cb);
virtual void getAndSet() const;
void doCallback(const std::string &data) const{ cb_ (path_, data); };
private:
CallbackType cb_;
};
class ChildrenWatch: public Watch
{
public:
typedef ChildrenWatchCallback CallbackType;
ChildrenWatch(ZooKeeper *zk, const std::string &path, const CallbackType &cb);
virtual void getAndSet() const;
void doCallback(const std::vector<std::string> &data) const { cb_ (path_, data); };
private:
CallbackType cb_;
};
//
class WatchPool
{
public:
template<class T>
WatchPtr createWatch(ZooKeeper *zk, const std::string &path, const typename T::CallbackType &cb)
{
std::string name = typeid(T).name() + path;
WatchMap::iterator itr = watchMap_.find(name);
if(watchMap_.end() == itr)
{
WatchPtr wp(new T(zk, path, cb));
watchMap_[name] = wp;
return wp;
}
else
{
return itr->second;
}
}
template<class T>
WatchPtr getWatch(const std::string &path)
{
std::string name = typeid(T).name() + path;
WatchMap::iterator itr = watchMap_.find(name);
if(watchMap_.end() == itr)
{
return WatchPtr();
}
else
{
return itr->second;
}
}
//
void getAndSetAll() const
{
for(WatchMap::const_iterator it = watchMap_.begin(); it != watchMap_.end(); ++it)
{
it->second->getAndSet();
}
}
private:
typedef std::map<std::string, WatchPtr> WatchMap;
WatchMap watchMap_;
};
//
static void dataCompletion(int rc, const char *value, int valueLen, const struct Stat *stat, const void *data);
static void stringsCompletion(int rc, const struct String_vector *strings, const void *data);
static void defaultWatcher(zhandle_t *zh, int type, int state, const char *path,void *watcherCtx);
//
ZkRet createTheNode(int flag, const std::string &path, const std::string &value, char *rpath, int rpathlen, bool recursive);
//
void miliSleep(int milisec);
//
zhandle_t *zhandle_;
std::string connectString_;
bool connected_;
ZooLogLevel defaultLogLevel_;
WatchPool watchPool_;
//
FILE *logStream_;
};
#endif
