C++中的Reactor原理與實(shí)現(xiàn)
更新時(shí)間：2022年07月01日 12:18:54 作者：恒者走天下
reactor設(shè)計(jì)模式是event-driven?architecture的一種實(shí)現(xiàn)方式，處理多個(gè)客戶端并發(fā)的向服務(wù)端請(qǐng)求服務(wù)的場(chǎng)景，每種服務(wù)在服務(wù)端可能由多個(gè)方法組成，這篇文章主要介紹了Reactor原理與實(shí)現(xiàn),需要的朋友可以參考下
一、Reactor介紹
reactor設(shè)計(jì)模式是event-driven architecture的一種實(shí)現(xiàn)方式，處理多個(gè)客戶端并發(fā)的向服務(wù)端請(qǐng)求服務(wù)的場(chǎng)景。每種服務(wù)在服務(wù)端可能由多個(gè)方法組成。reactor會(huì)解耦并發(fā)請(qǐng)求的服務(wù)并分發(fā)給對(duì)應(yīng)的事件處理器來(lái)處理。
中心思想是將所有要處理的I/o事件注冊(cè)到一個(gè)中心I/o多路復(fù)用器上，同時(shí)主線程/進(jìn)程阻塞在多路復(fù)用器上;一旦有I/o事件到來(lái)或是準(zhǔn)備就緒(文件描述符或socket可讀、寫)，多路復(fù)用器返回并將事先注冊(cè)的相應(yīng)l/o事件分發(fā)到對(duì)應(yīng)的處理器中。
處理機(jī)制為：主程序?qū)⑹录约皩?duì)應(yīng)事件處理的方法在Reactor上進(jìn)行注冊(cè), 如果相應(yīng)的事件發(fā)生，Reactor將會(huì)主動(dòng)調(diào)用事件注冊(cè)的接口，即 回調(diào)函數(shù).
二、代碼實(shí)現(xiàn)
前提準(zhǔn)備：1單例模式：?jiǎn)卫Ｊ?Singleton Pattern，也稱為單件模式)，使用最廣泛的設(shè)計(jì)模式之一。其意圖是保證一個(gè)類（結(jié)構(gòu)體）僅有一個(gè)實(shí)例，并提供一個(gè)訪問(wèn)它的全局訪問(wèn)點(diǎn)，該實(shí)例被所有程序模塊共享。
2.回調(diào)函數(shù)：把一段可執(zhí)行的代碼像參數(shù)傳遞那樣傳給其他代碼，而這段代碼會(huì)在某個(gè)時(shí)刻被調(diào)用執(zhí)行，這就叫做回調(diào)。
對(duì)epoll反應(yīng)堆中結(jié)構(gòu)體定義
/*fd包含的屬性*/
struct nitem { // fd
int fd; //要監(jiān)聽(tīng)的文件描述符
int status; //是否在監(jiān)聽(tīng)：1->在紅黑樹(shù)上（監(jiān)聽(tīng)），0->不在（不監(jiān)聽(tīng)）
int events; //對(duì)應(yīng)的監(jiān)聽(tīng)事件， EPOLLIN和EPOLLOUT（不同的事件，走不同的回調(diào)函數(shù)）
void *arg; //指向自己結(jié)構(gòu)體指針
#if 0
NCALLBACK callback;
#else
NCALLBACK *readcb; // epollin
NCALLBACK *writecb; // epollout
NCALLBACK *acceptcb; // epollin
#endif
unsigned char sbuffer[BUFFER_LENGTH]; //
int slength;
unsigned char rbuffer[BUFFER_LENGTH];
int rlength;
};
/*分塊存儲(chǔ)*/
struct itemblock {
struct itemblock *next;
struct nitem *items;
};
/*epoll反應(yīng)堆中包括通信的fd以及epoll的（epfd）*/
struct reactor {
int epfd;
struct itemblock *head;
};
單例模式，創(chuàng)建reactor的一個(gè)實(shí)例
/*單例模式*/
struct reactor *instance = NULL;
int init_reactor(struct reactor *r) {
if (r == NULL) return -1;
int epfd = epoll_create(1); //int size
r->epfd = epfd;
// fd --> item
r->head = (struct itemblock*)malloc(sizeof(struct itemblock));
if (r->head == NULL) {
close(epfd);
return -2;
}
memset(r->head, 0, sizeof(struct itemblock));
r->head->items = (struct nitem *)malloc(MAX_EPOLL_EVENT * sizeof(struct nitem));
if (r->head->items == NULL) {
free(r->head);
close(epfd);
return -2;
}
memset(r->head->items, 0, (MAX_EPOLL_EVENT * sizeof(struct nitem)));
r->head->next = NULL;
return 0;
}
struct reactor *getInstance(void) { //singleton
if (instance == NULL) {
instance = (struct reactor *)malloc(sizeof(struct reactor));
if (instance == NULL) return NULL;
memset(instance, 0, sizeof(struct reactor));
if (0 > init_reactor(instance)) {
free(instance);
return NULL;
}
}
return instance;
}
事件注冊(cè)
/*nreactor_set_event(listenfd, accept_callback, ACCEPT_CB, NULL);*/
/*nreactor_set_event(fd, read_callback, READ_CB, NULL);*/
/*fd找到對(duì)應(yīng)事件*/
/*驅(qū)動(dòng)注冊(cè)*/
int nreactor_set_event(int fd, NCALLBACK cb, int event, void *arg) {
struct reactor *r = getInstance();
struct epoll_event ev = {0};
//1
if (event == READ_CB) {
r->head->items[fd].fd = fd;
r->head->items[fd].readcb = cb;
r->head->items[fd].arg = arg;
ev.events = EPOLLIN;
}
//2
else if (event == WRITE_CB) {
r->head->items[fd].fd = fd;
r->head->items[fd].writecb = cb;
r->head->items[fd].arg = arg;
ev.events = EPOLLOUT;
}
//3
else if (event == ACCEPT_CB) {
r->head->items[fd].fd = fd;
r->head->items[fd].acceptcb = cb; //回調(diào)函數(shù)
r->head->items[fd].arg = arg;
ev.events = EPOLLIN;
}
ev.data.ptr = &r->head->items[fd];
/*NOSET_CB 0*/
if (r->head->items[fd].events == NOSET_CB) {
if (epoll_ctl(r->epfd, EPOLL_CTL_ADD, fd, &ev) < 0) {
printf("epoll_ctl EPOLL_CTL_ADD failed, %d\n", errno);
return -1;
}
r->head->items[fd].events = event;
} else if (r->head->items[fd].events != event) {
if (epoll_ctl(r->epfd, EPOLL_CTL_MOD, fd, &ev) < 0) {
printf("epoll_ctl EPOLL_CTL_MOD failed\n");
return -1;
}
r->head->items[fd].events = event;
}
return 0;
}
回調(diào)函數(shù)書寫
int write_callback(int fd, int event, void *arg) {
struct reactor *R = getInstance();
unsigned char *sbuffer = R->head->items[fd].sbuffer;
int length = R->head->items[fd].slength;
int ret = send(fd, sbuffer, length, 0);
if (ret < length) {
nreactor_set_event(fd, write_callback, WRITE_CB, NULL);
} else {
nreactor_set_event(fd, read_callback, READ_CB, NULL);
}
return 0;
}
// 5k qps
int read_callback(int fd, int event, void *arg) {
struct reactor *R = getInstance();
unsigned char *buffer = R->head->items[fd].rbuffer;
#if 0 //ET
int idx = 0, ret = 0;
while (idx < BUFFER_LENGTH) {
ret = recv(fd, buffer+idx, BUFFER_LENGTH-idx, 0);
if (ret == -1) {
break;
} else if (ret > 0) {
idx += ret;
} else {// == 0
break;
}
}
if (idx == BUFFER_LENGTH && ret != -1) {
nreactor_set_event(fd, read_callback, READ_CB, NULL);
} else if (ret == 0) {
nreactor_set_event
//close(fd);
} else {
nreactor_set_event(fd, write_callback, WRITE_CB, NULL);
}
#else //LT
int ret = recv(fd, buffer, BUFFER_LENGTH, 0);
if (ret == 0) { // fin
nreactor_del_event(fd, NULL, 0, NULL);
close(fd);
} else if (ret > 0) {
unsigned char *sbuffer = R->head->items[fd].sbuffer;
memcpy(sbuffer, buffer, ret);
R->head->items[fd].slength = ret;
printf("readcb: %s\n", sbuffer);
nreactor_set_event(fd, write_callback, WRITE_CB, NULL);
}
#endif
}
// web server
// ET / LT
int accept_callback(int fd, int event, void *arg) {
int connfd;
struct sockaddr_in client;
socklen_t len = sizeof(client);
if ((connfd = accept(fd, (struct sockaddr *)&client, &len)) == -1) {
printf("accept socket error: %s(errno: %d)\n", strerror(errno), errno);
return 0;
}
nreactor_set_event(connfd, read_callback, READ_CB, NULL);
}
監(jiān)聽(tīng)描述符變化
// accept --> EPOLL
/*epoll_wait監(jiān)聽(tīng)0*/
int reactor_loop(int listenfd) {
struct reactor *R = getInstance();
struct epoll_event events[POLL_SIZE] = {0};
while (1) {
int nready = epoll_wait(R->epfd, events, POLL_SIZE, -1);
if (nready == -1) {
continue;
}
int i = 0;
for (i = 0;i < nready;i ++) {
struct nitem *item = (struct nitem *)events[i].data.ptr;
int connfd = item->fd;
if (connfd == listenfd) { //
item->acceptcb(listenfd, 0, NULL);
} else {
if (events[i].events & EPOLLIN) { //
item->readcb(connfd, 0, NULL);
}
if (events[i].events & EPOLLOUT) {
item->writecb(connfd, 0, NULL);
}
}
}
}
return 0;
}
完整代碼實(shí)現(xiàn)
#define MAXLNE 4096
#define POLL_SIZE 1024
#define BUFFER_LENGTH 1024
#define MAX_EPOLL_EVENT 1024
#define NOSET_CB 0
#define READ_CB 1
#define WRITE_CB 2
#define ACCEPT_CB 3
/*單例模式*/
typedef int NCALLBACK(int fd, int event, void *arg);
/*fd包含的屬性*/
struct nitem { // fd
int fd; //要監(jiān)聽(tīng)的文件描述符
int status; //是否在監(jiān)聽(tīng)：1->在紅黑樹(shù)上（監(jiān)聽(tīng)），0->不在（不監(jiān)聽(tīng)）
int events; //對(duì)應(yīng)的監(jiān)聽(tīng)事件， EPOLLIN和EPOLLOUT（不同的事件，走不同的回調(diào)函數(shù)）
void *arg; //指向自己結(jié)構(gòu)體指針
#if 0
NCALLBACK callback;
#else
NCALLBACK *readcb; // epollin
NCALLBACK *writecb; // epollout
NCALLBACK *acceptcb; // epollin
#endif
unsigned char sbuffer[BUFFER_LENGTH]; //
int slength;
unsigned char rbuffer[BUFFER_LENGTH];
int rlength;
};
/*分塊存儲(chǔ)*/
struct itemblock {
struct itemblock *next;
struct nitem *items;
};
/*epoll反應(yīng)堆中包括通信的fd以及epoll的（epfd）*/
struct reactor {
int epfd;
struct itemblock *head;
};
/*初始化結(jié)構(gòu)體*/
int init_reactor(struct reactor *r);
int read_callback(int fd, int event, void *arg);
int write_callback(int fd, int event, void *arg);
int accept_callback(int fd, int event, void *arg);
/*單例模式*/
struct reactor *instance = NULL;
struct reactor *getInstance(void) { //singleton
if (instance == NULL) {
instance = (struct reactor *)malloc(sizeof(struct reactor));
if (instance == NULL) return NULL;
memset(instance, 0, sizeof(struct reactor));
if (0 > init_reactor(instance)) {
free(instance);
return NULL;
}
}
return instance;
}
/*nreactor_set_event(listenfd, accept_callback, ACCEPT_CB, NULL);*/
/*nreactor_set_event(fd, read_callback, READ_CB, NULL);*/
/*fd找到對(duì)應(yīng)事件*/
/*驅(qū)動(dòng)注冊(cè)*/
int nreactor_set_event(int fd, NCALLBACK cb, int event, void *arg) {
struct reactor *r = getInstance();
struct epoll_event ev = {0};
//1
if (event == READ_CB) {
r->head->items[fd].fd = fd;
r->head->items[fd].readcb = cb;
r->head->items[fd].arg = arg;
ev.events = EPOLLIN;
}
//2
else if (event == WRITE_CB) {
r->head->items[fd].fd = fd;
r->head->items[fd].writecb = cb;
r->head->items[fd].arg = arg;
ev.events = EPOLLOUT;
}
//3
else if (event == ACCEPT_CB) {
r->head->items[fd].fd = fd;
r->head->items[fd].acceptcb = cb; //回調(diào)函數(shù)
r->head->items[fd].arg = arg;
ev.events = EPOLLIN;
}
ev.data.ptr = &r->head->items[fd];
/*NOSET_CB 0*/
if (r->head->items[fd].events == NOSET_CB) {
if (epoll_ctl(r->epfd, EPOLL_CTL_ADD, fd, &ev) < 0) {
printf("epoll_ctl EPOLL_CTL_ADD failed, %d\n", errno);
return -1;
}
r->head->items[fd].events = event;
} else if (r->head->items[fd].events != event) {
if (epoll_ctl(r->epfd, EPOLL_CTL_MOD, fd, &ev) < 0) {
printf("epoll_ctl EPOLL_CTL_MOD failed\n");
return -1;
}
r->head->items[fd].events = event;
}
return 0;
}
/*nreactor_del_event(fd, NULL, 0, NULL);*/
/*下樹(shù)*/
/*nreactor_del_event(fd, NULL, 0, NULL);*/
int nreactor_del_event(int fd, NCALLBACK cb, int event, void *arg) {
struct reactor *r = getInstance();
struct epoll_event ev = {0};
ev.data.ptr = arg;
epoll_ctl(r->epfd, EPOLL_CTL_DEL, fd, &ev);
r->head->items[fd].events = 0;
return 0;
}
int write_callback(int fd, int event, void *arg) {
struct reactor *R = getInstance();
unsigned char *sbuffer = R->head->items[fd].sbuffer;
int length = R->head->items[fd].slength;
int ret = send(fd, sbuffer, length, 0);
if (ret < length) {
nreactor_set_event(fd, write_callback, WRITE_CB, NULL);
} else {
nreactor_set_event(fd, read_callback, READ_CB, NULL);
}
return 0;
}
// 5k qps
int read_callback(int fd, int event, void *arg) {
struct reactor *R = getInstance();
unsigned char *buffer = R->head->items[fd].rbuffer;
#if 0 //ET
int idx = 0, ret = 0;
while (idx < BUFFER_LENGTH) {
ret = recv(fd, buffer+idx, BUFFER_LENGTH-idx, 0);
if (ret == -1) {
break;
} else if (ret > 0) {
idx += ret;
} else {// == 0
break;
}
}
if (idx == BUFFER_LENGTH && ret != -1) {
nreactor_set_event(fd, read_callback, READ_CB, NULL);
} else if (ret == 0) {
nreactor_set_event
//close(fd);
} else {
nreactor_set_event(fd, write_callback, WRITE_CB, NULL);
}
#else //LT
int ret = recv(fd, buffer, BUFFER_LENGTH, 0);
if (ret == 0) { // fin
nreactor_del_event(fd, NULL, 0, NULL);
close(fd);
} else if (ret > 0) {
unsigned char *sbuffer = R->head->items[fd].sbuffer;
memcpy(sbuffer, buffer, ret);
R->head->items[fd].slength = ret;
printf("readcb: %s\n", sbuffer);
nreactor_set_event(fd, write_callback, WRITE_CB, NULL);
}
#endif
}
// web server
// ET / LT
int accept_callback(int fd, int event, void *arg) {
int connfd;
struct sockaddr_in client;
socklen_t len = sizeof(client);
if ((connfd = accept(fd, (struct sockaddr *)&client, &len)) == -1) {
printf("accept socket error: %s(errno: %d)\n", strerror(errno), errno);
return 0;
}
nreactor_set_event(connfd, read_callback, READ_CB, NULL);
}
int init_server(int port) {
int listenfd;
struct sockaddr_in servaddr;
char buff[MAXLNE];
if ((listenfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
printf("create socket error: %s(errno: %d)\n", strerror(errno), errno);
return 0;
}
memset(&servaddr, 0, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(port);
if (bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr)) == -1) {
printf("bind socket error: %s(errno: %d)\n", strerror(errno), errno);
return 0;
}
if (listen(listenfd, 10) == -1) {
printf("listen socket error: %s(errno: %d)\n", strerror(errno), errno);
return 0;
}
return listenfd;
}
int init_reactor(struct reactor *r) {
if (r == NULL) return -1;
int epfd = epoll_create(1); //int size
r->epfd = epfd;
// fd --> item
r->head = (struct itemblock*)malloc(sizeof(struct itemblock));
if (r->head == NULL) {
close(epfd);
return -2;
}
memset(r->head, 0, sizeof(struct itemblock));
r->head->items = (struct nitem *)malloc(MAX_EPOLL_EVENT * sizeof(struct nitem));
if (r->head->items == NULL) {
free(r->head);
close(epfd);
return -2;
}
memset(r->head->items, 0, (MAX_EPOLL_EVENT * sizeof(struct nitem)));
r->head->next = NULL;
return 0;
}
// accept --> EPOLL
/*epoll_wait監(jiān)聽(tīng)0*/
int reactor_loop(int listenfd) {
struct reactor *R = getInstance();
struct epoll_event events[POLL_SIZE] = {0};
while (1) {
int nready = epoll_wait(R->epfd, events, POLL_SIZE, -1);
if (nready == -1) {
continue;
}
int i = 0;
for (i = 0;i < nready;i ++) {
struct nitem *item = (struct nitem *)events[i].data.ptr;
int connfd = item->fd;
if (connfd == listenfd) { //
item->acceptcb(listenfd, 0, NULL);
} else {
if (events[i].events & EPOLLIN) { //
item->readcb(connfd, 0, NULL);
}
if (events[i].events & EPOLLOUT) {
item->writecb(connfd, 0, NULL);
}
}
}
}
return 0;
}
int main(int argc, char **argv)
{
int connfd, n;
int listenfd = init_server(9999);
nreactor_set_event(listenfd, accept_callback, ACCEPT_CB, NULL);
//nreactor_set_event(listenfd, accept_callback, read_callback, write_callback);
reactor_loop(listenfd);
return 0;
}
到此這篇關(guān)于Reactor原理與實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)Reactor原理內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
OpenCV實(shí)現(xiàn)圖像輪廓檢測(cè)以及外接矩形
這篇文章主要為大家詳細(xì)介紹了OpenCV實(shí)現(xiàn)圖像輪廓檢測(cè)以及外接矩形，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-01-01
這篇文章主要為大家詳細(xì)介紹了C++迷宮問(wèn)題的求解算法，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2020-03-03
最近工作中遇到了需求,要?jiǎng)h除字符串中的所有空格,就要篩選出空格字符,這篇文章主要給大家介紹了關(guān)于利用C語(yǔ)言刪除輸入字符串中的空格的相關(guān)資料,需要的朋友可以參考下2022-12-12
C語(yǔ)言 風(fēng)靡一時(shí)的黃金礦工游戲?qū)崿F(xiàn)流程詳解
《黃金礦工》是一款非常經(jīng)典的游戲。在游戲中，玩家通過(guò)不斷挖礦，獲取金子，最終能夠闖入下一關(guān)。在這個(gè)過(guò)程中，會(huì)不斷有巖石、煙霧、老鼠來(lái)?yè)v亂，甚至還會(huì)出現(xiàn)扛著炸藥包的小老鼠，玩家必須戰(zhàn)勝它們，才能進(jìn)入更深的礦坑2021-11-11
這篇文章主要為大家介紹了C++和C中const的區(qū)別，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來(lái)幫助2021-11-11
vector,map,list,queue的區(qū)別詳細(xì)解析
如果我們需要隨機(jī)訪問(wèn)一個(gè)容器則vector要比list好得多。如果我們已知要存儲(chǔ)元素的個(gè)數(shù)則vector 又是一個(gè)比list好的選擇。如果我們需要的不只是在容器兩端插入和刪除元素則list顯然要比vector好2013-09-09
C語(yǔ)言沒(méi)有運(yùn)行時(shí)庫(kù)，無(wú)法自動(dòng)壓縮使用中的內(nèi)存，縮小堆棧所需內(nèi)存空間。若只申請(qǐng)內(nèi)存，沒(méi)有釋放，勢(shì)必造成系統(tǒng)內(nèi)存不斷減少、丟失。長(zhǎng)時(shí)間的運(yùn)行，最終導(dǎo)致系統(tǒng)死機(jī)。文章闡述了C語(yǔ)言垃圾產(chǎn)生的原因，并從引用計(jì)數(shù)、標(biāo)記一清除算法兩方面提出如何實(shí)現(xiàn)C語(yǔ)言的垃圾回收。2019-02-02
C++實(shí)現(xiàn)動(dòng)態(tài)順序表
這篇文章主要為大家詳細(xì)介紹了C++實(shí)現(xiàn)動(dòng)態(tài)順序表，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2020-05-05
基于OpenCV?差分法實(shí)現(xiàn)綠葉識(shí)別
物體識(shí)別是圖像處理學(xué)在現(xiàn)實(shí)生活中較多的應(yīng)用之一，本文提供了一種相對(duì)簡(jiǎn)單的思路來(lái)實(shí)現(xiàn)綠葉識(shí)別，適合初學(xué)圖像處理的新人研究參考。感興趣的同學(xué)可以關(guān)注一下2021-11-11
C++實(shí)現(xiàn)矩陣原地轉(zhuǎn)置算法
這篇文章主要介紹了C++實(shí)現(xiàn)矩陣原地轉(zhuǎn)置算法,非常經(jīng)典的算法,需要的朋友可以參考下2014-08-08