2.3.1（项目）kv存储——框架梳理(待定)
一、过一遍代码路线：（1）底层网络框架；（2）业务层的协议筹划(发什么数据，以及收什么数据)；（3）业务层的引擎筹划；
1. 底层网络框架：
目标：通过定义函数创建网络层和业务层的联系，并从原有已以网络框架为主改造后以业务框架为main函数，确保此后网络层隔离和透明（以reactor.c为例）。
要求：reactor.c(手写)、ntyco协程(看懂)、io_uring(看懂)。
做法：用做reactor的做法来做kvstore的网络框架
核心：是设置自定义的布局体，明白什么变乱执行什么回调函数。while(1){}循环里，（a）EPOLLIN变乱执行recv_callback，EPOLLOUT变乱执行send_callback。（b）EPOLLIN变乱包罗sockfd和clientfd，分别执行accept函数和recv/send函数，可以通过recv_callback的index加以区分。
要点：
知识点（1）：通过宏定义开启或关闭不同的功能，由此方便新功能编写。
#define ENABLE_KVSTORE 1 #if ENABLE_KVSTORE
kvs_request(&connect_list);
#endif
#if ENABLE_KVSTORE
kvs_response(&connect_list);
#endif
知识点（2）：(1) 协议处理 → 放到kvstore.c；（2）main入口函数 → 放到网络框架(reactor.c)里面？还是放到kvstore.c里面?
答：如果把main入口函数放到网络框架(reactor.c)就表明主要步调由网络框架实现，按照这个逻辑kvstore.c只是其中的一个业务，而流程不应该如许。所以main要放到kvstore.c里面，以保证网络层被完全的隔离。对于一个服务而言，我们要终极封装成：网络层仅必要传输端口(port)和协议的接口(kvs_protocol)这两个端口已往，
知识点（3）：网络框架(reactor.c)里怎样与kvstore.c毗连：
（a）定义kvs_request和kvs_response；
（b）send_cb和recv_cb分别设置kvs_request和kvs_response；
（c）send_cb里的kvs_request使用kvs_handler处理数据 ←
kvs_handler通过“static msg_handler kvs_handler” 定义范例，在网络框架(reactor.c)主函数里被handler赋值←
handler作为网络框架(reactor.c)主函数里其中一个输入参数，被msg_handler布局体赋值 ←
布局体msg_handler在网络框架(reactor.c)被定义，包罗*msg、length、*response，同时也在kvstore.h文件里。←
作为在网络框架(reactor.c)主函数输入参数的handler，在kvstore.c里被定义为kvs_protocal。网络框架(reactor.c)的命令行 “static msg_handler kvs_handler” 为了 “kvs_handler = handler”。←
网络层的编译命令
https://i-blog.csdnimg.cn/direct/fd6aae8155b14fa6a488275a264076c9.png
此处的代码：
server.h(基本没变化)
#ifndef __SERVER_H__
#define __SERVER_H__
#define BUFFER_LENGTH 128
#define ENABLE_KVSTORE 1
typedef int(*RCALLBACK)(int fd);
struct conn {
int fd;
char rbuffer;
int rlength;
char wbuffer;
int wlength;
RCALLBACKsend_callback;
RCALLBACKrecv_callback;
};
#if ENABLE_KVSTORE
int kvs_request(struct conn *c);
int kvs_response(struct conn *c);
#endif
#endif
kvs_reactor.c
#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <string.h>
#include <errno.h>
#include <sys/epoll.h>
#include <unistd.h>
#include <sys/time.h>
#include "server.h"
//#define BUFFER_LENGTH 1024
#define CONNECT_SIZE 1048576
#define PORT_SIZE 1
#define TIME_SUB_MS(tv1, tv2)((tv1.tv_sec - tv2.tv_sec) * 1000 + (tv1.tv_usec - tv2.tv_usec) / 1000)
#if ENABLE_KVSTORE
typedef int (*msg_handler)(char *msg, int length, char *response);
static msg_handler kvs_handler;
int kvs_request(struct conn *c){
// printf("recv %d: %s\n", c->rlength ,c->rbuffer);
c->wlength = kvs_handler(c->rbuffer, c->rlength, c->wbuffer);
}
#endif
int kvs_response(struct conn *c){
}
//typedef int(*RCALLBACK)(int fd);
int accept_cb(int fd);
int recv_cb(int fd);
int send_cb(int fd);
int epfd = 0;
struct timeval begin;
/*
struct conn {
int fd;
char rbuffer;
int rlength;
char wbuffer;
int wlength;
RCALLBACKsend_callback;
RCALLBACKrecv_callback;
};
*/
struct conn connect_list = {0};
int set_event(int fd, int event, int flag){
if (flag == 1){ // non-zero add
struct epoll_event ev;
ev.data.fd = fd;
ev.events = event;
epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);
} else { // zero MOD
struct epoll_event ev;
ev.data.fd = fd;
ev.events = event;
epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);
}
}
int event_register(int fd, int event){
if (fd < 0){
return -1;
}
connect_list.fd = fd;
connect_list.recv_callback = recv_cb;
connect_list.send_callback = send_cb;
memset(connect_list.rbuffer, 0, BUFFER_LENGTH);
connect_list.rlength = 0;
memset(connect_list.wbuffer, 0, BUFFER_LENGTH);
connect_list.wlength = 0;
set_event(fd, EPOLLIN, 1);
}
int accept_cb(int fd){
struct sockaddr_in clientaddr;
socklen_t len = sizeof(clientaddr);
int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len);
if (clientfd < 0){
printf("accept error:%d, %s\n", errno, strerror(errno));
return -1;
}
event_register(clientfd, EPOLLIN);
if((clientfd % 1000) == 0){
struct timeval current;
gettimeofday(¤t, NULL);
int time_used = TIME_SUB_MS(current, begin);
memcpy(&begin, ¤t, sizeof(struct timeval));
printf("Accept clientfd : %d, time_used: %d\n", clientfd, time_used);
}
return 0;
}
int recv_cb(int fd){
int count = recv(fd, connect_list.rbuffer, BUFFER_LENGTH, 0);
if (count ==0){
printf("client disconnect: %d\n", fd);
close(fd);
epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);// unfinished
return 0;
}
connect_list.rlength = count;
// printf("rbuffer: %s, recv: %d\n", connect_list.rbuffer, count);
# if 0 //echo
connect_list.wlength = connect_list.rlength;
memcpy(connect_list.wbuffer, connect_list.rbuffer, connect_list.wlength);
#endif
#if ENABLE_KVSTORE
kvs_request(&connect_list);
#endif
set_event(fd, EPOLLOUT, 0);
return count;
}
int send_cb(int fd){
#if ENABLE_KVSTORE
kvs_response(&connect_list);
#endif
int count = 0;
count = send(fd, connect_list.wbuffer, connect_list.wlength, 0);
// printf("wbuffer: %s, send: %d\n", connect_list.wbuffer, count);
set_event(fd, EPOLLIN, 0);
return count;
}
int r_init_server(unsigned short port){
int sockfd = socket(AF_INET, SOCK_STREAM, 0);
struct sockaddr_in servaddr;
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr= htonl(INADDR_ANY);// uint32_t
servaddr.sin_port = htons(port);
if (-1 == bind(sockfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr))) {
printf("bind failed: %s\n", strerror(errno));
}
listen(sockfd, 10);
// printf("listen finished: %d\n", sockfd);
return sockfd;
}
int reactor_start(unsigned short port, msg_handler handler){
epfd = epoll_create(1);
//unsigned short port = 2000;
kvs_handler =handler;
int i = 0;
for (i = 0; i < PORT_SIZE; i ++){
int sockfd = r_init_server(port + i);
connect_list.fd =sockfd;
connect_list.recv_callback = accept_cb;
printf("sockfd: %d\n", sockfd);
set_event(sockfd, EPOLLIN, 1);
}
gettimeofday(&begin, NULL);
while(1){
struct epoll_event events = {0};
int nready = epoll_wait(epfd, events, 1024, -1);
// printf("epoll_wait is %d\n", nready);
int i = 0;
for (i = 0; i < nready; i ++){
int connfd = events.data.fd;
if (events.events & EPOLLIN){
connect_list.recv_callback(connfd);
}
if (events.events & EPOLLOUT){
connect_list.send_callback(connfd);
}
}
}
return 0;
} kvstore.c
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include "kvstore.h"
/*
* msg: request message
* length: length of request message
* response:need to seed
*@return: length of response
*/
int kvs_protocal(char *msg, int length, char *response){
printf("recv %d: %s\n", length, msg);
memcpy(response, msg, length);
return strlen(response);
}
int main(int argc, char *argv[]){ //input parameters
if (argc != 2) return -1;
// ./program Hello World: Number of arguments is 3
int port = atoi(argv);
#if (NETWORK_SELECT==NETWORK_REACTOR)
reactor_start(port, kvs_protocal);
#elif
#endif
} kvstore.h
#ifndef __KV_STORE_H__
#define__KV_STORE_H__
#define NETWORK_REACTOR 0
#define NETWORK_SELECT NETWORK_REACTOR
typedef int (*msg_handler)(char *msg, int length, char *response);
extern int reactor_start(unsigned short port, msg_handler handler);
const char *command[] = {
"SET", "GET", "DEL", "MOD", "EXIST"
};
const char *response[] = {
};
#endif
2. 业务层的协议筹划：
目标：在前面已经创建tcp公共协议毗连的根本上，创建发什么数据和接收什么数据的私有协议 / 自定义协议 / 发数据的格式。
具体做法：对于网络部门已经实现了数据的接收和发送，所以接下来我们要在收数据的地方修改，在原有reactor.c的根本上，
（1）新建kvstore.c定义有关业务联系的函数(接收request和输出response)；
（2）新建kvstore.h为函数(接收request和输出response)定义函数具体的协议、具体response。
具体5个response格式：
command(命令)
SET Key Value
GET Key
DEL Key
MOD Key Value
EXIST Key （3）在上一部门网络已经封装根本上，以kvstore.c里的自定义kvs_protocol函数(也就是对应网络框架(reactor.c) 的主函数第2个输入参数)为入口函数，进行具体的协议解析功能(也就是定义接收和发送的数据规则寄义，如下2种模式)。
SET Key Value
Get Key 要点：
知识点（1）：针对tcp的分包和粘包，问题1是什么是分包和粘包，问题2是怎么解决这个问题？
答1：由于send(fd, buffer, length, 0)的次数和recv(fd, buffer, length, 0)大概不一致，大概send()多次发送recv一次性都接收了(这就是粘包)，所以要区分出每次的数据(分包解数据)。
答2：分包的时间必要和对方约定好格式，同时tcp毗连满足假设(1) recv()先接收的数据就是对方先发送的数据，保证了接收数据的次序；(2) 数据在传输过程中不会发生丢包情况，保证了数据必达性。
方式1. send()在发包时就确定约定的长度，比如接收2次就可以接收完整的数据包，第一次接收2字节作为标志，第二次再接收数据包。
SEND:
SET Key Value\r\n // 15表示15个字符 第一次：接收开头的2个字符，同时通过这2个字符的值得知后面还要接收15字符
short length = 0;
recv(fd, &length, 2, 0);
第二次：接收剩余数据15个字符
recv(fd, buffer, length, 0); 方式2. 仿照Redis，先确定有多少(N)个token后就先发送字符N，具体发送格式如下：
Refis → SET Key Value
发送顺序如下
3\r\n // 3表示“SET Key Value”一共是3个token(单词)
3\r\n // 3表示第一个token(单词)“SET"是3个字符的长度
SET\r\n // 发送一个token，“SET"
3\r\n // 3表示第二个token(单词)“Key"也是3个字符的长度
Key\r\n // 发送一个token，“Key"
5\r\n // 5表示第三个token(单词)“Value"也是5个字符的长度
Value\r\n // 发送一个token，“Value" 伪代码如下，终极要读(1+2*3)=7次
自行实现readline()函数，要求每一行读到"\r\n"就停止
count = atoi(buffer);
for (i = 0; i < count; i ++){
readline(fd, tokenlen);
readline(fd, token);
}
3. 业务层的引擎筹划
数据布局用红黑树、希哈、数组、链表(手写)
二、备注：
(1) “const char *command[] ={};” 什么意思？以及为什么是const以及为什么*?
答：const: 确保字符串内容不能被修改。*: 表现指针，每个数组元素都指向一个字符串。
(2) 定义函数不消分号";"，而布局体必要，那么什么时间必要加“;”？
答：定义一个范例（比方 struct 或 typedef）告诉编译器这个范例存在但并不提供执行行为，必要以分号结束。
（a）必要分号的地方：变量声明、布局体定义、typedef 定义新范例
int a = 10; // 表示变量声明结束
struct Point {
int x;
char y;
}; // 分号表示 `struct Point` 的定义结束
typedef struct {
int x;
char y;
} PointType; // 分号表示 `typedef` 的结束
（b）不必要分号的地方：函数定义、#define 宏(宏是简单的文本替换)。
int add(int a, int b) {
return a + b; // 函数体内语句的分号只是代码逻辑的一部分
}
#define SQUARE(x) ((x) * (x)) // 宏定义
（3）为什么分为 .c 和 .h 文件?
答：每个 .c 文件负责一个模块的实现，多个模块通过 .h 文件互相调用。.h 文件只能声明接口，不能包罗具体实现。自定义头文件使用双引号：#include "xxxx.h"。
（4）"int main(int argc, char *argv[]){}" 是一个标准的 C 语言步调入口函数。
答：argc表现参数个数 (Argument Count)，argv表现参数向量 (Argument Vector)。用于获取命令行传递给步调的参数。
int main(int argc, char *argv[]) {
printf("Number of arguments: %d\n", argc);
for (int i = 0; i < argc; i++) {
printf("Argument %d: %s\n", i, argv);
}
return 0;
}
运行：
./program Hello World
输出：
Number of arguments: 3
Argument 0: ./program
Argument 1: Hello
Argument 2: World
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]