从零开始实现一个C++高性能服务器框架----配置模块
Posted johnsonli99
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了从零开始实现一个C++高性能服务器框架----配置模块相关的知识，希望对你有一定的参考价值。
此项目是根据sylar框架实现，是从零开始重写sylar，也是对sylar丰富与完善
项目地址：https://gitee.com/lzhiqiang1999/server-framework
简介
项目介绍：实现了一个基于协程的服务器框架，支持多线程、多协程协同调度；支持以异步处理的方式提高服务器性能；封装了网络相关的模块，包括socket、http、servlet等，支持快速搭建HTTP服务器或WebSokcet服务器。
详细内容：日志模块，使用宏实现流式输出，支持同步日志与异步日志、自定义日志格式、日志级别、多日志分离等功能。线程模块，封装pthread相关方法，封装常用的锁包括（信号量，读写锁，自旋锁等）。IO协程调度模块，基于ucontext_t实现非对称协程模型，以线程池的方式实现多线程，多协程协同调度，同时依赖epoll实现了事件监听机制。定时器模块，使用最小堆管理定时器，配合IO协程调度模块可以完成基于协程的定时任务调度。hook模块，将同步的系统调用封装成异步操作（accept, recv, send等），配合IO协程调度能够极大的提升服务器性能。Http模块，封装了sokcet常用方法，支持http协议解析，客户端实现连接池发送请求，服务器端实现servlet模式处理客户端请求，支持单Reator多线程，多Reator多线程模式的服务器。
配置模块
1. 主要功能
- 支持yaml格式的配置文件解析
- 使用模板完成基础类型，复杂类型（vector、map、set等），自定义类型的序列化与反序列化
- 利用回调机制，在加载配置时，完成配置的更新
- 使用yaml-cpp库，实现配置文件读取
- 约定大于配置
2. 功能演示
- 配置文件
test.yaml
tcp:
connect:
timeout: 10000
// 约定的配置
ConfigVar<int>::ptr tcp_config = Config::Lookup("tcp.connect.timeout", 5000, "tcp connect timeout");
LOG_INFO(LOG_ROOT()) << "before: " << tcp_config->getValue(); // before: 5000
// 读取配置文件
YAML::Node root = YAML::LoadFile("test.yaml");
// 加载配置文件，并更改配置值
Config::LoadFromYaml(root);
LOG_INFO(LOG_ROOT()) << "after: " << tcp_config->getValue(); // after: 10000
3. 模块介绍
3.1 ConfigVarBase
- 配置基本信息。基类。
class ConfigVarBase
public:
typedef std::shared_ptr<ConfigVarBase> ptr;
ConfigVarBase(const std::string& name, const std::string& description = "")
:m_name(name)
,m_description(description)
//把名字都转换成小写
std::transform(m_name.begin(), m_name.end(), m_name.begin(), ::tolower);
virtual ~ConfigVarBase()
//get, set方法
virtual std::string toString() = 0; // 序列化，转换成string
virtual bool fromString(const std::string& val) = 0; /
protected:
std::string m_name; /// 配置的字段的名字
std::string m_description; /// 字段的描述
;
3.2 ConfigVar
- 配置变量。使用模板的方式完成基本类型，复杂类型（vector、map、set等），自定义类型的序列化和反序列化
template<class T, class FromStr = LexicalCast<std::string, T>
, class ToStr = LexicalCast<T, std::string>>
class ConfigVar : public ConfigVarBase
public:
typedef RWMutex RWMutexType;
typedef std::shared_ptr<ConfigVar> ptr;
typedef std::function<void (const T& old_value, const T& new_value)> on_change_cb;
std::string toString() override;
bool fromString(const std::string& val) override;
uint64_t addListener(on_change_cb cb);
private:
T m_val; /// 配置变量
std::map<uint64_t, on_change_cb> m_cbs; /// 回调函数，当配置有改变时，回调
RWMutexType m_mutex; /// 读写锁，读多写少
- 使用模板实现序列化与反序列化
- ConfigVar 是一个模板类
template<class T, class FromStr = LexicalCast<std::string, T>, class ToStr = LexicalCast<T, std::string>>，其中FromStr完成序列化（string to value）；ToStr完成反序列化（value to string）。 - 基本类型。F类型转为T类型
/** * @brief 类型转换模板类(F 源类型, T 目标类型) */ template<class F, class T> class LexicalCast public: /** * @brief 类型转换 * @param[in] v 源类型值 * @return 返回v转换后的目标类型 * @exception 当类型不可转换时抛出异常 */ T operator()(const F& v) return boost::lexical_cast<T>(v); ;- vector的序列化与反序列化。
/** * @brief 类型转换模板类片特化(YAML String 转换成 std::vector<T>) */ template<class T> class LexicalCast<std::string, std::vector<T> > public: std::vector<T> operator()(const std::string& v) //"yaml string [1,2,3] 转换成yaml node" YAML::Node node = YAML::Load(v); typename std::vector<T> vec; std::stringstream ss; for (size_t i = 0; i < node.size(); ++i) ss.str(""); ss << node[i]; vec.push_back(LexicalCast<std::string, T>()(ss.str())); return vec; ; /** * @brief 类型转换模板类片特化(std::vector<T> 转换成 YAML String) */ template<class T> class LexicalCast<std::vector<T>, std::string> public: std::string operator()(const std::vector<T>& v) YAML::Node node(YAML::NodeType::Sequence); for (auto& i : v) node.push_back(YAML::Load(LexicalCast<T, std::string>()(i))); std::stringstream ss; ss << node; return ss.str(); ;- map的序列化与反序列化。
/** * @brief 类型转换模板类片特化(YAML String 转换成 std::map<std::string, T>) */ template<class T> class LexicalCast<std::string, std::map<std::string, T> > public: std::map<std::string, T> operator()(const std::string& v) YAML::Node node = YAML::Load(v); typename std::map<std::string, T> vec; std::stringstream ss; for (auto it = node.begin(); it != node.end(); ++it) ss.str(""); ss << it->second; vec.insert(std::make_pair(it->first.Scalar(), LexicalCast<std::string, T>()(ss.str()))); return vec; ; /** * @brief 类型转换模板类片特化(std::map<std::string, T> 转换成 YAML String) */ template<class T> class LexicalCast<std::map<std::string, T>, std::string> public: std::string operator()(const std::map<std::string, T>& v) YAML::Node node(YAML::NodeType::Map); for (auto& i : v) node[i.first] = YAML::Load(LexicalCast<T, std::string>()(i.second)); std::stringstream ss; ss << node; return ss.str(); ;- 自定义类型的序列化与反序列化将在后文讲解
- ConfigVar 是一个模板类
- ConfigVar如何完成序列化与反序列化
- ConfigVar的模板
template<class T, class FromStr = LexicalCast<std::string, T>, class ToStr = LexicalCast<T, std::string>>。这里使用了模板偏特化，如果T是普通类型（int），会自动匹配到LexicalCast<std::string, int>；如果T是复杂类型（vector<int>）,会自动匹配到LexicalCast<std::string, vector<int>>。 - 仿函数实现序列化
//T to string YAML string std::string toString() override try RWMutexType::ReadLock lock(m_mutex); return ToStr()(m_val); // ToStr()(m_val)是仿函数，实际上是因为我们重写了operator() catch (std::exception& e) LOG_ERROR(LOG_ROOT()) << "ConfigVar::toString exception" << e.what() << " convert: " << typeid(m_val).name() << " to string"; return "";- 仿函数实现反序列化
//YAML string to T bool fromString(const std::string& val) override try setValue(FromStr()(val)); // FromStr()(val)是仿函数，实际上是因为我们重写了operator() return true; catch (std::exception& e) LOG_ERROR(LOG_ROOT()) << "ConfigVar::fromString exception" << e.what() << " convert: string to" << typeid(m_val).name(); return false; - ConfigVar的模板
- 回调机制。加载时完成配置更新。当
fromString()时，会调用setValue()，此时会根据新值和旧值，判断是否需要调用回调，完成配置更新
typedef std::function<void (const T& old_value, const T& new_value)> on_change_cb; // 回调函数的类型
std::map<uint64_t, on_change_cb> m_cbs; /// 该配置的所有回调
// 添加监听器
uint64_t addListener(on_change_cb cb)
//每次由系统决定key值，再返回，用于delListener
static uint64_t s_idx = 0;
RWMutexType::WriteLock lock(m_mutex);
++s_idx;
m_cbs[s_idx] = cb;
return s_idx;
// 设置配置值
void setValue(const T& value)
RWMutexType::ReadLock lock(m_mutex);
if (value == m_val) return;
//只有当old 和 new不同时才回调
for (auto& i : m_cbs)
i.second(m_val, value);
RWMutexType::WriteLock lock(m_mutex);
m_val = value;
- 这里使用了读写锁保证线程安全，因为配置模块主要是读多写少，能保证最大效率
3.3 Config
- 配置管理。负责管理所有的配置信息
static ConfigVarMap s_datas
//管理类
class Config
public:
typedef RWMutex RWMutexType;
typedef std::map<std::string, ConfigVarBase::ptr> ConfigVarMap;
//查找一个配置变量，没找到就创建一个新的，放到s_datas
//name：A.B
template<class T>
static typename ConfigVar<T>::ptr Lookup(const std::string& name, const T& default_value, const std::string& description = "");
//查找name
template<class T>
static typename ConfigVar<T>::ptr Lookup(const std::string& name);
//解析yaml文件，读取配置信息
static void LoadFromYaml(const YAML::Node& root);
//查看当前配置表的信息
static void Visit(std::function<void(johnsonli::ConfigVarBase::ptr)> cb);
private:
//查找name，返回ConfigVarBase::ptr
static ConfigVarBase::ptr LookupBase(const std::string& name);
private:
/**
* @brief 返回所有的配置项
*/
static ConfigVarMap& GetDatas()
static ConfigVarMap s_datas;
return s_datas;
//如果是静态成员变量，可能对象会先创建出来，但是静态变量还没有初始化
static RWMutexType& GetMutex()
static RWMutexType s_mutex;
return s_mutex;
;
- 自定义类型的序列化与反序列化
- 由于需要完成对日志的配置，因此需要完成对日志器的序列化与反序列化。这里我们需要先抽取出两个结构体
struct LogAppenderDefine与struct LogDefine。
//LogAppender struct LogAppenderDefine int type = 0; //1 File, 2 Stdout LogLevel::Level level = LogLevel::UNKNOW; std::string formatter; std::string file; bool operator==(const LogAppenderDefine& oth) const return type == oth.type && level == oth.level && formatter == oth.formatter && file == oth.file; ; //Logger struct LogDefine std::string name; LogLevel::Level level = LogLevel::UNKNOW; std::vector<LogAppenderDefine> appenders; bool operator==(const LogDefine& oth) const return name == oth.name && level == oth.level && appenders == oth.appenders; bool operator<(const LogDefine& oth) const return name < oth.name; ;struct LogDefine表示日志器的配置信息，因此需要完成日志器的序列化与反序列化
//LogDefine偏特化 YAML string to LogDefine template<> class LexicalCast<std::string, LogDefine> public: LogDefine operator()(const std::string& str)... ; //LogDefine偏特化 LogDefine to YAML string template<> class LexicalCast<LogDefine, std::string> public: std::string operator()(const LogDefine& i) ... ;- 在加载日志配置时，我们添加了回调函数，实现在加载时就完成配置更改
这里使用了一个方法，使得添加回调函数的执行在main函数之前。在类的构造函数中处理，声明静态变量，这样就会在main函数之前进行构造，完成处理。ConfigVar<std::set<LogDefine>>::ptr g_log_defines = Config::Lookup("logs", std::set<LogDefine>(), "logs config"); // LoadFromYaml（YAML string to T） --> 在SetValue中观察者模式触发日志更改事件 --> 调用回调函数(添加、修改、删除) struct LogIniter LogIniter() g_log_defines->addListener([] (const std::set<LogDefine>& old_val, const std::set<LogDefine>& new_val) // 添加，new_val有，old_val没有 // 修改，new_val有，old_val有，但不相同 // 删除，new_val没有，old_val有 ; static LogIniter __log_init;- 日志配置的格式
logs: - name: root level: debug appenders: - type: FileLogAppender level: debug file: root.log formatter: "%d%TF%T[%p]%T[%c]%T%f:%l%T%m%n" - type: StdoutLogAppender level: debug formatter: "%d%Y-%m-%d %H:%M:%S%T%t%T%N%T%F%T[%p]%T[%c]%T%f:%l%T%m%n" - name: system level: debug appenders: - type: FileLogAppender level: debug file: root.log formatter: "%d%TF%T[%p]%T[%c]%T%f:%l%T%m%n" - type: StdoutLogAppender level: debug formatter: "%d%Y-%m-%d %H:%M:%S%T%t%T%N%T%F%T[%p]%T[%c]%T%f:%l%T%m%n" - 由于需要完成对日志的配置，因此需要完成对日志器的序列化与反序列化。这里我们需要先抽取出两个结构体
- 下图是整个加载Logger配置的过程
从零开始，徒手撸一个简单的 RPC 框架，轻松搞定！
得知了RPC（远程过程调用）简单来说就是调用远程的服务就像调用本地方法一样，其中用到的知识有序列化和反序列化、动态代理、网络传输、动态加载、反射这些知识点。发现这些知识都了解一些。所以就想着试试自己实现一个简单的RPC框架，即巩固了基础的知识，也能更加深入的了解RPC原理。当然一个完整的RPC框架包含了许多的功能，例如服务的发现与治理，网关等等。本篇只是简单的实现了一个调用的过程。
传参出参分析
一个简单请求可以抽象为两步
那么就根据这两步进行分析，在请求之前我们应该发送给服务端什么信息？而服务端处理完以后应该返回客户端什么信息？
1、在请求之前我们应该发送给服务端什么信息？
由于我们在客户端调用的是服务端提供的接口，所以我们需要将客户端调用的信息传输过去，那么我们可以将要传输的信息分为两类
-
第一类是服务端可以根据这个信息找到相应的接口实现类和方法 -
第二类是调用此方法传输的参数信息
那么我们就根据要传输的两类信息进行分析，什么信息能够找到相应的实现类的相应的方法？要找到方法必须要先找到类，这里我们可以简单的用Spring提供的Bean实例管理ApplicationContext进行类的寻找。所以要找到类的实例只需要知道此类的名字就行，找到了类的实例，那么如何找到方法呢？在反射中通过反射能够根据方法名和参数类型从而找到这个方法。那么此时第一类的信息我们就明了了，那么就建立相应的是实体类存储这些信息。
@Data
public class Request implements Serializable {
private static final long serialVersionUID = 3933918042687238629L;
private String className;
private String methodName;
private Class<?> [] parameTypes;
private Object [] parameters;
}
2、服务端处理完以后应该返回客户端什么信息？
上面我们分析了客户端应该传输什么信息给服务端，那么服务端处理完以后应该传什么样的返回值呢？这里我们只考虑最简单的情况，客户端请求的线程也会一直在等着，不会有异步处理这一说，所以这么分析的话就简单了，直接将得到的处理结果返回就行了。
@Data
public class Response implements Serializable {
private static final long serialVersionUID = -2393333111247658778L;
private Object result;
}
由于都涉及到了网络传输，所以都要实现序列化的接口
如何获得传参信息并执行？-客户端
上面我们分析了客户端向服务端发送的信息都有哪些？那么我们如何获得这些信息呢？首先我们调用的是接口，所以我们需要写自定义注解然后在程序启动的时候将这些信息加载在Spring容器中。有了这些信息那么我们就需要传输了，调用接口但是实际上执行的确实网络传输的过程，所以我们需要动态代理。那么就可以分为以下两步
-
初始化信息阶段：将key为接口名，value为动态接口类注册进Spring容器中 -
执行阶段：通过动态代理，实际执行网络传输
1、初始化信息阶段
由于我们使用Spring作为Bean的管理，所以要将接口和对应的代理类注册进Spring容器中。而我们如何找到我们想要调用的接口类呢？我们可以自定义注解进行扫描。将想要调用的接口全部注册进容器中。
创建一个注解类，用于标注哪些接口是可以进行Rpc的
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface RpcClient {
}
@RpcClient注解的扫描类
RpcInitConfig，将其注册进Spring容器中
public class RpcInitConfig implements ImportBeanDefinitionRegistrar{
@Override
public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
ClassPathScanningCandidateComponentProvider provider = getScanner();
//设置扫描器
provider.addIncludeFilter(new AnnotationTypeFilter(RpcClient.class));
//扫描此包下的所有带有@RpcClient的注解的类
Set<BeanDefinition> beanDefinitionSet = provider.findCandidateComponents("com.example.rpcclient.client");
for (BeanDefinition beanDefinition : beanDefinitionSet){
if (beanDefinition instanceof AnnotatedBeanDefinition){
//获得注解上的参数信息
AnnotatedBeanDefinition annotatedBeanDefinition = (AnnotatedBeanDefinition) beanDefinition;
String beanClassAllName = beanDefinition.getBeanClassName();
Map<String, Object> paraMap = annotatedBeanDefinition.getMetadata()
.getAnnotationAttributes(RpcClient.class.getCanonicalName());
//将RpcClient的工厂类注册进去
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(RpcClinetFactoryBean.class);
//设置RpcClinetFactoryBean工厂类中的构造函数的值
builder.addConstructorArgValue(beanClassAllName);
builder.getBeanDefinition().setAutowireMode(AbstractBeanDefinition.AUTOWIRE_BY_TYPE);
//将其注册进容器中
registry.registerBeanDefinition(
beanClassAllName ,
builder.getBeanDefinition());
}
}
}
//允许Spring扫描接口上的注解
protected ClassPathScanningCandidateComponentProvider getScanner() {
return new ClassPathScanningCandidateComponentProvider(false) {
@Override
protected boolean isCandidateComponent(AnnotatedBeanDefinition beanDefinition) {
return beanDefinition.getMetadata().isInterface() && beanDefinition.getMetadata().isIndependent();
}
};
}
}
由于上面注册的是工厂类，所以我们建立一个工厂类RpcClinetFactoryBean继承Spring中的FactoryBean类，由其统一创建@RpcClient注解的代理类
如果对FactoryBean类不了解的可以参见FactoryBean讲解
@Data
public class RpcClinetFactoryBean implements FactoryBean {
@Autowired
private RpcDynamicPro rpcDynamicPro;
private Class<?> classType;
public RpcClinetFactoryBean(Class<?> classType) {
this.classType = classType;
}
@Override
public Object getObject(){
ClassLoader classLoader = classType.getClassLoader();
Object object = Proxy.newProxyInstance(classLoader,new Class<?>[]{classType},rpcDynamicPro);
return object;
}
@Override
public Class<?> getObjectType() {
return this.classType;
}
@Override
public boolean isSingleton() {
return false;
}
}
注意此处的
getObjectType方法，在将工厂类注入到容器中的时候，这个方法返回的是什么Class类型那么注册进容器中就是什么Class类型。
然后看一下我们创建的代理类rpcDynamicPro
@Component
@Slf4j
public class RpcDynamicPro implements InvocationHandler {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String requestJson = objectToJson(method,args);
Socket client = new Socket("127.0.0.1", 20006);
client.setSoTimeout(10000);
//获取Socket的输出流，用来发送数据到服务端
PrintStream out = new PrintStream(client.getOutputStream());
//获取Socket的输入流，用来接收从服务端发送过来的数据
BufferedReader buf = new BufferedReader(new InputStreamReader(client.getInputStream()));
//发送数据到服务端
out.println(requestJson);
Response response = new Response();
Gson gson =new Gson();
try{
//从服务器端接收数据有个时间限制（系统自设，也可以自己设置），超过了这个时间，便会抛出该异常
String responsJson = buf.readLine();
response = gson.fromJson(responsJson, Response.class);
}catch(SocketTimeoutException e){
log.info("Time out, No response");
}
if(client != null){
//如果构造函数建立起了连接，则关闭套接字，如果没有建立起连接，自然不用关闭
client.close(); //只关闭socket，其关联的输入输出流也会被关闭
}
return response.getResult();
}
public String objectToJson(Method method,Object [] args){
Request request = new Request();
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
String className = method.getDeclaringClass().getName();
request.setMethodName(methodName);
request.setParameTypes(parameterTypes);
request.setParameters(args);
request.setClassName(getClassName(className));
GsonBuilder gsonBuilder = new GsonBuilder();
gsonBuilder.registerTypeAdapterFactory(new ClassTypeAdapterFactory());
Gson gson = gsonBuilder.create();
return gson.toJson(request);
}
private String getClassName(String beanClassName){
String className = beanClassName.substring(beanClassName.lastIndexOf(".")+1);
className = className.substring(0,1).toLowerCase() + className.substring(1);
return className;
}
}
我们的客户端已经写完了，传给服务端的信息我们也已经拼装完毕了。剩下的工作就简单了，开始编写服务端的代码。
服务端处理完以后应该返回客户端什么信息？-服务端
服务端的代码相比较客户端来说要简单一些。可以简单分为下面三步
-
拿到接口名以后，通过接口名找到实现类 -
通过反射进行对应方法的执行 -
返回执行完的信息
那么我们就根据这三步进行编写代码
1、拿到接口名以后，通过接口名找到实现类
如何通过接口名拿到对应接口的实现类呢？这就需要我们在服务端启动的时候将其对应信息加载进去
@Component
@Log4j
public class InitRpcConfig implements CommandLineRunner {
@Autowired
private ApplicationContext applicationContext;
public static Map<String,Object> rpcServiceMap = new HashMap<>();
@Override
public void run(String... args) throws Exception {
Map<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(Service.class);
for (Object bean: beansWithAnnotation.values()){
Class<?> clazz = bean.getClass();
Class<?>[] interfaces = clazz.getInterfaces();
for (Class<?> inter : interfaces){
rpcServiceMap.put(getClassName(inter.getName()),bean);
log.info("已经加载的服务:"+inter.getName());
}
}
}
private String getClassName(String beanClassName){
String className = beanClassName.substring(beanClassName.lastIndexOf(".")+1);
className = className.substring(0,1).toLowerCase() + className.substring(1);
return className;
}
}
rpcServiceMap存储的就是接口名和其对应的实现类的对应关系。
2、通过反射进行对应方法的执行
此时拿到了对应关系以后就能根据客户端传过来的信息找到相应的实现类中的方法。然后进行执行并返回信息就行
public Response invokeMethod(Request request){
String className = request.getClassName();
String methodName = request.getMethodName();
Object[] parameters = request.getParameters();
Class<?>[] parameTypes = request.getParameTypes();
Object o = InitRpcConfig.rpcServiceMap.get(className);
Response response = new Response();
try {
Method method = o.getClass().getDeclaredMethod(methodName, parameTypes);
Object invokeMethod = method.invoke(o, parameters);
response.setResult(invokeMethod);
} catch (NoSuchMethodException e) {
log.info("没有找到"+methodName);
} catch (IllegalAccessException e) {
log.info("执行错误"+parameters);
} catch (InvocationTargetException e) {
log.info("执行错误"+parameters);
}
return response;
}
现在我们两个服务都启动起来并且在客户端进行调用就发现只是调用接口就能调用过来了。
总结
到现在一个简单的RPC就完成了，但是其中还有很多的功能需要完善，例如一个完整RPC框架肯定还需要服务注册与发现，而且双方通信肯定也不能是直接开启一个线程一直在等着，肯定需要是异步的等等的各种功能。后面随着学习的深入，这个框架也会慢慢增加一些东西。不仅是对所学知识的一个应用，更是一个总结。有时候学一个东西学起来觉得很简单，但是真正应用的时候就会发现各种各样的小问题。比如在写这个例子的时候碰到一个问题就是@Autowired的时候一直找不到SendMessage的类型，最后才发现是工厂类RpcClinetFactoryBean中的getObjectType中的返回类型写错了，我之前写的是
public Class<?> getObjectType() {
return this.getClass();;
}
这样的话注册进容器的就是RpcClinetFactoryBean类型的而不是SendMessage的类型。
juejin.cn/post/6844903764445364232
以上是关于从零开始实现一个C++高性能服务器框架----配置模块的主要内容，如果未能解决你的问题，请参考以下文章
从零开始实现一个C++高性能服务器框架----Socket模块