C++中的Reactor原理与实现

2024-12-04 00:55:03

一、Reactor介绍

reactor设计模式是event-driven architecture的一种实现方式，处理多个客户端并发的向服务端请求服务的场景。每种服务在服务端可能由多个方法组成。reactor会解耦并发请求的服务并分发给对应的事件处理器来处理。

中心思想是将所有要处理的I/o事件注册到一个中心I/o多路复用器上，同时主线程/进程阻塞在多路复用器上;一旦有I/o事件到来或是准备就绪(文件描述符或socket可读、写)，多路复用器返回并将事先注册的相应l/o事件分发到对应的处理器中。

处理机制为：主程序将事件以及对应事件处理的方法在Reactor上进行注册, 如果相应的事件发生，Reactor将会主动调用事件注册的接口，即回调函数.

二、代码实现

前提准备：1单例模式：单例模式(Singleton Pattern，也称为单件模式)，使用最广泛的设计模式之一。其意图是保证一个类（结构体）仅有一个实例，并提供一个访问它的全局访问点，该实例被所有程序模块共享。
2.回调函数：把一段可执行的代码像参数传递那样传给其他代码，而这段代码会在某个时刻被调用执行，这就叫做回调。

对epoll反应堆中结构体定义

/*fd包含的属性*/
struct nitem { // fd

	int fd;		//要监听的文件描述符

	int status;	//是否在监听：1->在红黑树上（监听），0->不在（不监听）
	int events;	//对应的监听事件，	EPOLLIN和EPOLLOUT（不同的事件，走不同的回调函数）
	void *arg;	//指向自己结构体指针
#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;

};

/*分块存储*/
struct itemblock {

	struct itemblock *next;
	struct nitem *items;

};
/*epoll反应堆中包括通信的fd以及epoll的（epfd）*/
struct reactor {

	int epfd;
	struct itemblock *head; 

};

单例模式，创建reactor的一个实例

/*单例模式*/
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;
}

事件注册

/*nreactor_set_event(listenfd, accept_callback, ACCEPT_CB, NULL);*/
/*nreactor_set_event(fd, read_callback, READ_CB, NULL);*/
/*fd找到对应事件*/
/*驱动注册*/
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;	//回调函数
		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;
}

回调函数书写

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);
}

监听描述符变化

// accept --> EPOLL
/*epoll_wait监听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;
}

完整代码实现

#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;		//要监听的文件描述符
	int status;	//是否在监听：1->在红黑树上（监听），0->不在（不监听）
	int events;	//对应的监听事件，	EPOLLIN和EPOLLOUT（不同的事件，走不同的回调函数）
	void *arg;	//指向自己结构体指针
#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;

};
/*分块存储*/
struct itemblock {
	struct itemblock *next;
	struct nitem *items;
};
/*epoll反应堆中包括通信的fd以及epoll的（epfd）*/
struct reactor {
	int epfd;
	struct itemblock *head;
};
/*初始化结构体*/
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找到对应事件*/
/*驱动注册*/
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;	//回调函数
		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);*/
/*下树*/
/*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监听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;
}

到此这篇关于Reactor原理与实现的文章就介绍到这了,更多相关Reactor原理内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！

如何使用Reactor完成类似Flink的操作

一.背景 Flink在处理流式任务的时候有很大的优势,其中windows等操作符可以很方便的完成聚合任务,但是Flink是一套独立的服务,业务流程中如果想使用需要将数据发到kafka,用Flink处理完再发到kafka,然后再做业务处理,流程很繁琐. 比如在业务代码中想要实现类似Flink的window按时间批量聚合功能,如果纯手动写代码比较繁琐,使用Flink又太重,这种场景下使用响应式编程RxJava.Reactor等的window.buffer操作符可以很方便的实现. 响应式编程框架也早已
Reactor反应器的实现方法详解

大多数应用都会使用ACE_Reactor::instance()提供的默认反应器实例.但是你也可以选择自己的反应器,这是因为ACE使用了Bridge模式(使用两个不同的类:一个是编程接口,另一个是实现,第一个类会把各个操作传给第二个类).例如使用线程池反应器实现:ACE_TP_Reactor* tp_reactor = new ACE_TP_Reactor;ACE_Reactor* my_reactor = new ACE_Reactor(tp_reactor, 1);//1表示my_react
gateway、webflux、reactor-netty请求日志输出方式

目录 gateway.webflux.reactor-netty请求日志输出场景思路解决方案 spring-webflux.gateway.springboot-start-web问题 Spring-webflux Spring-gateway gateway.webflux.reactor-netty请求日志输出场景在使用spring cloud gateway时想要输出请求日志,考虑到两种实现方案方案一官网中使用Reactor Netty Access Logs方案,配置“-D
ACE反应器(Reactor)模式的深入分析

反应器(Reactor):用于事件多路分离和分派的体系结构模式通常的,对一个文件描述符指定的文件或设备, 有两种工作方式: 阻塞与非阻塞.所谓阻塞方式的意思是指, 当试图对该文件描述符进行读写时, 如果当时没有东西可读,或者暂时不可写, 程序就进入等待状态, 直到有东西可读或者可写为止.而对于非阻塞状态, 如果没有东西可读, 或者不可写, 读写函数马上返回, 而不会等待. 在前面的章节中提到的Tcp通信的例子中,就是采用的阻塞式的工作方式:当接收tcp数据时,如果远端没有数据可以读,则会一直阻塞
C++中的Reactor原理与实现

目录一.Reactor介绍二.代码实现一.Reactor介绍 reactor设计模式是event-driven architecture的一种实现方式,处理多个客户端并发的向服务端请求服务的场景.每种服务在服务端可能由多个方法组成.reactor会解耦并发请求的服务并分发给对应的事件处理器来处理. 中心思想是将所有要处理的I/o事件注册到一个中心I/o多路复用器上,同时主线程/进程阻塞在多路复用器上;一旦有I/o事件到来或是准备就绪(文件描述符或socket可读.写),多路复用器返回并将事
epoll封装reactor原理剖析示例详解

目录 reactor是什么? reactor模型三个重要组件与流程分析组件流程将epoll封装成reactor事件驱动封装每一个连接sockfd变成ntyevent 封装epfd和ntyevent变成ntyreactor 封装读.写.接收连接等事件对应的操作变成callback 给每个客户端的ntyevent设置属性将ntyevent加入到epoll中由内核监听将ntyevent从epoll中去除读事件回调函数写事件回调函数接受新连接事件回调函数 reactor运行 react
详解JSP 中Spring工作原理及其作用

详解JSP 中Spring工作原理及其作用 1.springmvc请所有的请求都提交给DispatcherServlet,它会委托应用系统的其他模块负责负责对请求进行真正的处理工作. 2.DispatcherServlet查询一个或多个HandlerMapping,找到处理请求的Controller. 3.DispatcherServlet请请求提交到目标Controller 4.Controller进行业务逻辑处理后,会返回一个ModelAndView 5.Dispathcher查询一个或多个
Python中的浮点数原理与运算分析

本文实例讲述了Python中的浮点数原理与运算.分享给大家供大家参考,具体如下: 先看一个违反直觉的例子: >>> s = 0. >>> for i in range(10): s += .1 >>> s 0.9999999999999999 # 错误被累加再看一个更为普遍,直接影响判断逻辑的例子: >>> from math import sqrt >>> a = sqrt(2) >>> a*a
ucenter中词语过滤原理分析

本文分析了ucenter中词语过滤原理.分享给大家供大家参考,具体如下: 过滤词语表: id admin find replacement findpattern 1 UCenterAdminist 访问大 /访问/is 2 UCenterAdminist 4655 45 /4655/is 3 UCenterAdminist fdsaf dfsa /fdsaf/is 4 UCenterAdminist 有机会在 /有机会/is 组建缓存数据: //private function _get_
Yii2中Restful API原理实例分析

本文实例分析了Yii2中Restful API原理.分享给大家供大家参考,具体如下: Yii2 有个很重要的特性是对 Restful API的默认支持, 通过短短的几个配置就可以实现简单的对现有Model的RESTful API 这里通过分析rest部分源码,简单剖析下yii2 实现 restful 的原理,并通过一些定制实现对关联模型的RESTful api 操作. ~ 代表 extends from 的关系 | | rest/ | | |-Action.php ~ `\yii\base\
C#中foreach实现原理详解

本文主要记录我在学习C#中foreach遍历原理的心得体会. 对集合中的要素进行遍历是所有编码中经常涉及到的操作,因此大部分编程语言都把此过程写进了语法中,比如C#中的foreach.经常会看到下面的遍历代码: var lstStr = new List<string> { "a", "b" }; foreach (var str in lstStr) { Console.WriteLine(str); } 实际此代码的执行过程: var lstStr
浅谈SSH框架中spring的原理

在ssh项目中,是有明确分工的,spring的作用就相当于将struts和hibernate连接起来,是将两个没有关系的框架的特性,方法,action都放在spring的配置文件中使他们建立关系.取他门各自所长.而这些做法他们自己不知道,他们是听命于spring调度的,他的的任务只是做好自己的事情. 这样做的好处就是任务结构分明,struts只管理显示与做什么,hibernate只关心怎么做,而spring就相当于领导,所以一切的类都要交给spring的工厂创建,这是一种良好的开发模式,体现了一
angular2中Http请求原理与用法详解

本文实例讲述了angular2中Http请求原理与用法.分享给大家供大家参考,具体如下: 提供HTTP服务 HttpModule并不是Angular的核心模块. 它是Angular用来进行Web访问的一种可选方式,并位于一个名叫@angular/http的独立附属模块中. 编辑app.module.ts import { HttpModule, JsonpModule } from '@angular/http'; @NgModule({ imports: [ HttpModule, Jsonp
Android6.0开发中屏幕旋转原理与流程分析

本文实例讲述了Android6.0开发中屏幕旋转原理与流程.分享给大家供大家参考,具体如下: 从Android 系统开发开始,这里写下Android 6.0 屏幕旋转系统分析. 第一部分 Kenel Android 系统屏幕旋转得以实现,是靠从底层驱动gsensor 中获取数据,从而判断屏幕方向的.kernel sensor的驱动先就不在这里赘述,简单介绍下,gsensor 驱动注册input 事件在/dev/input/下,可以通过adb getevent -p 可以查看系统所有的输入事件.

C++中的Reactor原理与实现

目录

一、Reactor介绍

二、代码实现

相关推荐

随机推荐