从零开始实现一个C++高性能服务器框架----Socket模块
Posted johnsonli99
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了从零开始实现一个C++高性能服务器框架----Socket模块相关的知识，希望对你有一定的参考价值。
此项目是根据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多线程模式的服务器。
Socket模块
1. 主要功能
- 对Linux下socket相关方法的封装，包括bind、listen、connect、read/write系列等方法。
- 支持快速创建TCP、UDP对应的Socket。
2. 功能演示
- 模拟一个请求百度的客户端，并打印出响应
IPAddress::ptr addr = Address::LookupAnyIPAddress("www.baidu.com:80");
// 创建socket
Socket::ptr socket = Socket::CreateTCP(addr);
// 连接
socket->connect(addr);
//发送数据
const char buf[] = "GET / HTTP/1.1\\r\\n\\r\\n";
int rt = socket->send(buf, sizeof(buf));
if(rt <= 0)
LOG_INFO(g_logger) << "send fail";
return;
//接收数据
std::string buffers;
buffers.resize(4096);
rt = socket->recv(&buffers[0], 4096);
if(rt <= 0)
LOG_INFO(g_logger) << "recv fail";
return;
LOG_INFO(g_logger) << buffers;
3. 模块介绍
3.1 Socket
- 对socket相关方法的封装，包括以下内容
- 创建各种类型的套接字对象的方法（TCP套接字，UDP套接字，Unix域套接字）
- 设置套接字选项，比如超时参数
- bind/connect/listen方法，实现绑定地址、发起连接、发起监听功能
- accept方法，返回连入的套接字对象
- 发送、接收数据的方法
- 获取本地地址、远端地址的方法
- 获取套接字类型、地址类型、协议类型的方法
- 取消套接字读、写的方法
class Socket : public std::enable_shared_from_this<Socket>, Noncopyable
public:
typedef std::shared_ptr<Socket> ptr;
typedef std::weak_ptr<Socket> weak_ptr;
// 创建TCP Socket(满足地址类型)
static Socket::ptr CreateTCP(johnsonli::Address::ptr address);
// 创建UDP Socket(满足地址类型)
static Socket::ptr CreateUDP(johnsonli::Address::ptr address);
// 创建IPv4的TCP Socket
static Socket::ptr CreateTCPSocket();
// 创建IPv4的UDP Socket
static Socket::ptr CreateUDPSocket();
// 创建IPv6的TCP Socket
static Socket::ptr CreateTCPSocket6();
// 创建IPv6的UDP Socket
static Socket::ptr CreateUDPSocket6();
Socket(int family, int type, int protocol = 0);
virtual ~Socket();
int64_t getSendTimeout(); // 获取发送超时时间(毫秒)
void setSendTimeout(int64_t v); // 设置发送超时时间(毫秒)
int64_t getRecvTimeout(); // 获取接受超时时间(毫秒)
void setRecvTimeout(int64_t v); // 设置接受超时时间(毫秒)
// 获取sockopt @see getsockopt
bool getOption(int level, int option, void* result, socklen_t* len);
// 获取sockopt模板 @see getsockopt
template<class T>
bool getOption(int level, int option, T& result)
socklen_t length = sizeof(T);
return getOption(level, option, &result, &length);
// 设置sockopt @see setsockopt
bool setOption(int level, int option, const void* result, socklen_t len);
// 设置sockopt模板 @see setsockopt
template<class T>
bool setOption(int level, int option, const T& value)
return setOption(level, option, &value, sizeof(T));
/**
* @brief 接收connect链接
* @return 成功返回新连接的socket,失败返回nullptr
* @pre Socket必须 bind , listen 成功
*/
virtual Socket::ptr accept();
/**
* @brief 绑定地址
* @param[in] addr 地址
* @return 是否绑定成功
*/
virtual bool bind(const Address::ptr addr);
/**
* @brief 连接地址
* @param[in] addr 目标地址
* @param[in] timeout_ms 超时时间(毫秒)
*/
virtual bool connect(const Address::ptr addr, uint64_t timeout_ms = -1);
virtual bool reconnect(uint64_t timeout_ms = -1);
/**
* @brief 监听socket
* @param[in] backlog 未完成连接队列的最大长度
* @result 返回监听是否成功
* @pre 必须先 bind 成功
*/
virtual bool listen(int backlog = SOMAXCONN);
/**
* @brief 关闭socket
*/
virtual bool close();
/**
* @brief 发送数据
* @param[in] buffer 待发送数据的内存
* @param[in] length 待发送数据的长度
* @param[in] flags 标志字
* @return
* @retval >0 发送成功对应大小的数据
* @retval =0 socket被关闭
* @retval <0 socket出错
*/
virtual int send(const void* buffer, size_t length, int flags = 0);
/**
* @brief 发送数据
* @param[in] buffers 待发送数据的内存(iovec数组)
* @param[in] length 待发送数据的长度(iovec长度)
* @param[in] flags 标志字
* @return
* @retval >0 发送成功对应大小的数据
* @retval =0 socket被关闭
* @retval <0 socket出错
*/
virtual int send(const iovec* buffers, size_t length, int flags = 0);
/**
* @brief 发送数据
* @param[in] buffer 待发送数据的内存
* @param[in] length 待发送数据的长度
* @param[in] to 发送的目标地址
* @param[in] flags 标志字
* @return
* @retval >0 发送成功对应大小的数据
* @retval =0 socket被关闭
* @retval <0 socket出错
*/
virtual int sendTo(const void* buffer, size_t length, const Address::ptr to, int flags = 0);
/**
* @brief 发送数据
* @param[in] buffers 待发送数据的内存(iovec数组)
* @param[in] length 待发送数据的长度(iovec长度)
* @param[in] to 发送的目标地址
* @param[in] flags 标志字
* @return
* @retval >0 发送成功对应大小的数据
* @retval =0 socket被关闭
* @retval <0 socket出错
*/
virtual int sendTo(const iovec* buffers, size_t length, const Address::ptr to, int flags = 0);
/**
* @brief 接受数据
* @param[out] buffer 接收数据的内存
* @param[in] length 接收数据的内存大小
* @param[in] flags 标志字
* @return
* @retval >0 接收到对应大小的数据
* @retval =0 socket被关闭
* @retval <0 socket出错
*/
virtual int recv(void* buffer, size_t length, int flags = 0);
/**
* @brief 接受数据
* @param[out] buffers 接收数据的内存(iovec数组)
* @param[in] length 接收数据的内存大小(iovec数组长度)
* @param[in] flags 标志字
* @return
* @retval >0 接收到对应大小的数据
* @retval =0 socket被关闭
* @retval <0 socket出错
*/
virtual int recv(iovec* buffers, size_t length, int flags = 0);
/**
* @brief 接受数据
* @param[out] buffer 接收数据的内存
* @param[in] length 接收数据的内存大小
* @param[out] from 发送端地址
* @param[in] flags 标志字
* @return
* @retval >0 接收到对应大小的数据
* @retval =0 socket被关闭
* @retval <0 socket出错
*/
virtual int recvFrom(void* buffer, size_t length, Address::ptr from, int flags = 0);
/**
* @brief 接受数据
* @param[out] buffers 接收数据的内存(iovec数组)
* @param[in] length 接收数据的内存大小(iovec数组长度)
* @param[out] from 发送端地址
* @param[in] flags 标志字
* @return
* @retval >0 接收到对应大小的数据
* @retval =0 socket被关闭
* @retval <0 socket出错
*/
virtual int recvFrom(iovec* buffers, size_t length, Address::ptr from, int flags = 0);
// 输出信息到流中
virtual std::ostream& dump(std::ostream& os) const;
virtual std::string toString() const;
bool cancelRead(); // 取消读
bool cancelWrite(); // 取消写
bool cancelAccept(); // 取消accept
bool cancelAll(); // 取消所有事件
protected:
void initSock(); // 设置socket属性
void newSock(); // 创建socket m_sockfd = socket()
virtual bool init(int sock); // 初始化sock，调用initSock
protected:
int m_sockfd; /// socket句柄
int m_family; /// 协议簇
int m_type; /// 类型
int m_protocol; /// 协议
bool m_isConnected; /// 是否连接
Address::ptr m_localAddress; /// 本地地址
Address::ptr m_remoteAddress; /// 远端地址
;
// 流式输出socket
std::ostream& operator<<(std::ostream& os, const Socket& sock);
从零开始，徒手撸一个简单的 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++高性能服务器框架----Socket模块的主要内容，如果未能解决你的问题，请参考以下文章
从零开始实现一个C++高性能服务器框架----Socket模块