• reactor的原理与实现

    目录

    1.  Reactor模型

    2.  Reactor 模型有三个重要的组件：

    3.  reactor所需要结构体分析

    4.  epoll对fd操作的封装

    5.  reactor的操作

    6.  事件的回调函数

     7.  完整代码和测试结果

    总结

    1.  Reactor模型

    普通函数调用的机制：

            程序调用某函数，函数执行，程序等待，函数将结果和控制权返回给程序，程序继续处理。

            

    什么是Reactor?

            Reactor释义“反应堆”，框架：是一种事件驱动的反应堆模式,高效的事件处理模型。和普通函数调用的不同之处在于：应用程序不是主动的调用某个API完成处理，而是恰恰相反，Reactor逆置了事件处理流程，应用程序需要提供相应的接口并注册到Reactor上，如果相应的时间发生，Reactor将主动调用应用程序注册的接口，这些接口又称为“回调函数”。

            reactor反应堆：当有事件，事件类型可能不相同，需要提前注册好不同的事件处理函数。由epoll_wait获取同时到来的多个事件，并且根据数据的不同类型将事件分发给事件处理机制(事件处理器)，需要提前注册的哪些接口函数。

     

    2.  Reactor 模型有三个重要的组件：

    1.多路复用器：由操作系统提供，在 linux 上一般是 select, poll, epoll 等系统调用。

    2. 事件分发器：将多路复用器中返回的就绪事件分到对应的处理函数中。

     

    3. 事件处理器：负责处理特定事件的处理函数。

    具体流程如下：

    1. 注册读就绪事件和相应的事件处理器；

    2. 事件分离器等待事件；

    3. 事件到来，激活分离器，分离器调用事件对应的处理器；

    4. 事件处理器完成实际的读操作，处理读到的数据，注册新的事件，然后返还控制

    权。

    3.  reactor所需要结构体分析

    1. 管理每一个fd的结构体

    1. //管理每一个io fd的结构体
    2. struct ntyevent{
    3. 	int fd; //io fd
    4. 	int events;
    5. 	void *arg;
    6. 	int (*callback)(int fd, int events, void* arg); //执行回调函数
    7.  
    8. 	int status;	//判断是否已有事件
    9. 	char buffer[BUFFER_LENGTH]; //用户缓冲区
    10. 	int length; //用户缓冲区长度
    11. };

     2. 管理ntyevent fd的块

    1. //管理ntyevent fd的块
    2. struct eventblock{
    3. 	struct eventblock* next; //指向ntyevent fd集合
    4. 	struct ntyevent* events; //指向下一个ntyevent fd的块
    5. };

     3. Reactor结构体

     

    1. //reacotr结点
    2. struct ntyreactor{
    3. 	int epfd;	//epoll fd
    4. 	int blkcnt;	//ntyevent fd的块 计数
    5.  
    6. 	struct eventblock* evblks;	//指向ntyevent fd的块头结点
    7. };

     

    4.  epoll对fd操作的封装

    1. //io fd结构体设置
    2. void nty_event_set(struct ntyevent* ev, int fd, NCALLBACK callback, void* arg)
    3. {
    4. 	ev->fd = fd;
    5. 	ev->callback = callback;
    6. 	ev->events = 0;
    7. 	ev->arg = arg;
    8.  
    9. 	return ;
    10. }
    11.  
    12. //io fd add 事件
    13. int nty_event_add(int epfd, int events, struct ntyevent *ev)
    14. {
    15. 	struct epoll_event ep_ev = {0, {0}};
    16. 	ep_ev.data.ptr = ev;				//io fd结构体
    17. 	ep_ev.events = ev->events = events; //需要检测的fd事件
    18.  
    19. 	int op; //操作类型
    20. 	if(ev->status == 1){
    21. 		op = EPOLL_CTL_MOD; //修改
    22. 	}else{
    23. 		op = EPOLL_CTL_ADD; //添加
    24. 		ev->status = 1;		//标志已经添加
    25. 	}
    26. 		
    27. 	if(epoll_ctl(epfd, op, ev->fd, &ep_ev) <0 ){ //绑定
    28. 		printf("event add failed [fd=%d], events[%d]\n", ev->fd, events);
    29. 		return -1;
    30. 	}
    31.  
    32. 	return 0;
    33. }
    34.  
    35. //io fd del 事件
    36. int nty_event_del(int epfd, struct ntyevent* ev)
    37. {
    38. 	struct epoll_event ep_ev = {0, {0}};
    39. 	if(ev->status != 1){ //没有添加过检测的fd事件
    40. 		return -1;
    41. 	}
    42.  
    43. 	ep_ev.data.ptr = ev;
    44. 	ev->status = 0; //标志未添加
    45. 	epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, &ep_ev);
    46. 	return 0;
    47. }

     

    5.  reactor的操作

    1. //reactor扩展大小
    2. int ntyreactor_alloc(struct ntyreactor* reactor)
    3. {
    4. 	if(reactor == NULL) return -1;
    5. 	if(reactor->evblks == NULL) return -1;
    6.  
    7. 	struct eventblock* blk = reactor->evblks; //块的头结点
    8.  
    9. 	//找尾节点
    10. 	while(blk->next != NULL){  //找到尾节点
    11. 		blk = blk->next;
    12. 	}
    13.  
    14. 	struct ntyevent* evs = (struct ntyevent*)malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    15. 	if (evs == NULL) {
    16. 		printf("ntyreactor_alloc ntyevent failed\n");
    17. 		return -2;
    18. 	}
    19. 	memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    20.  
    21. 	struct eventblock *block = malloc(sizeof(struct eventblock));
    22. 	if (block == NULL) {
    23. 		printf("ntyreactor_alloc eventblock failed\n");
    24. 		return -3;
    25. 	}
    26. 	//io fd集合连接成块
    27. 	block->events = evs;
    28. 	block->next = NULL;
    29.  
    30. 	//指向新块
    31. 	blk->next = block;
    32. 	reactor->blkcnt ++;
    33.  
    34. 	return 0;
    35. }
    36.  
    37. //根据io fd来找fd结构体
    38. struct ntyevent *ntyreactor_idx(struct ntyreactor *reactor, int sockfd) {
    39. 	if (reactor == NULL) return NULL;
    40. 	if (reactor->evblks == NULL) return NULL;
    41.  
    42. 	int blkidx = sockfd / MAX_EPOLL_EVENTS; //在哪一个块
    43. 	while (blkidx >= reactor->blkcnt) {		//大小不够扩容
    44. 		ntyreactor_alloc(reactor);
    45. 	}
    46.  
    47. 	int i = 0;
    48. 	struct eventblock *blk = reactor->evblks; //头结点块
    49. 	while (i++ != blkidx && blk != NULL) {    //找到所在的块
    50. 		blk = blk->next;
    51. 	}
    52.  
    53. 	return &blk->events[sockfd % MAX_EPOLL_EVENTS]; //返回fd结构体
    54. }
    55.  
    56. //reactor初始化
    57. int ntyreactor_init(struct ntyreactor* reactor)
    58. {
    59. 	if(reactor == NULL) return -1;
    60. 	memset(reactor, 0, sizeof(struct ntyreactor));
    61.  
    62. 	reactor->epfd = epoll_create(1); 
    63. 	if (reactor->epfd <= 0) {
    64. 		printf("create epfd in %s err %s\n", __func__, strerror(errno));
    65. 		return -2;
    66. 	}
    67.  
    68. 	//创建第一个块
    69. 	struct ntyevent* evs = (struct ntyevent*)malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    70. 	if (evs == NULL) {
    71. 		printf("create epfd in %s err %s\n", __func__, strerror(errno));
    72. 		close(reactor->epfd);
    73. 		return -3;
    74. 	}
    75. 	memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    76.  
    77. 	struct eventblock *block = malloc(sizeof(struct eventblock));
    78. 	if (block == NULL) {
    79. 		free(evs);
    80. 		close(reactor->epfd);
    81. 		return -3;
    82. 	}
    83. 	block->events = evs;
    84. 	block->next = NULL;
    85.  
    86. 	reactor->evblks = block;
    87. 	reactor->blkcnt = 1;
    88.  
    89. 	return 0;
    90. }
    91.  
    92. //销毁reactor
    93. int ntyreactor_destory(struct ntyreactor* reactor)
    94. {
    95. 	close(reactor->epfd);
    96.  
    97. 	struct eventblock *blk = reactor->evblks;
    98. 	struct eventblock *blk_next;
    99. 	while (blk != NULL) {
    100. 		blk_next = blk->next;
    101.  
    102. 		free(blk->events);
    103. 		free(blk);
    104. 		
    105. 		blk = blk_next;
    106. 	}
    107.  
    108. 	return 0;
    109. }

    6.  事件的回调函数

    1. //recv回调
    2. int recv_cb(int fd, int events, void* arg)
    3. {
    4. 	struct ntyreactor* reactor = (struct ntyreactor*)arg;
    5. 	struct ntyevent* ev = ntyreactor_idx(reactor, fd);	
    6.  
    7. 	if(ev == NULL)return -1;
    8.  
    9. 	int len = recv(fd, ev->buffer, BUFFER_LENGTH, 0);
    10. 	nty_event_del(reactor->epfd, ev);
    11.  
    12. 	if (len > 0) {
    13. 		
    14. 		ev->length = len;
    15. 		ev->buffer[len] = '\0';
    16.  
    17. 		printf("recv [%d]:%s\n", fd, ev->buffer);
    18.  
    19. 		//将fd 设置为发送事件
    20. 		nty_event_set(ev, fd, send_cb, reactor);
    21. 		nty_event_add(reactor->epfd, EPOLLOUT, ev);
    22. 		
    23. 		
    24. 	} else if (len == 0) {  //客户端断开连接
    25.  
    26. 		nty_event_del(reactor->epfd, ev);
    27. 		printf("recv_cb --> disconnect\n");
    28. 		close(ev->fd);
    29. 		 
    30. 	} else { //返回错误
    31.  
    32. 		if (errno == EAGAIN && errno == EWOULDBLOCK) { //
    33. 			
    34. 		} else if (errno == ECONNRESET){
    35. 			nty_event_del(reactor->epfd, ev);
    36. 			close(ev->fd);
    37. 		}
    38. 		printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));
    39. 		
    40. 	}
    41.  
    42. 	return len;
    43. }
    44.  
    45.  
    46.  
    47. //send回调
    48. int send_cb(int fd, int events, void* arg)
    49. {
    50. 	struct ntyreactor* reactor = (struct ntyreactor*)arg;
    51. 	struct ntyevent* ev = ntyreactor_idx(reactor, fd);
    52.  
    53. 	if (ev == NULL) return -1;
    54.  
    55. 	int len = send(fd, ev->buffer, ev->length, 0);
    56. 	if (len > 0) {
    57. 		printf("send[fd=%d], [%d]%s\n", fd, len, ev->buffer);
    58.  
    59. 		//发送后，将fd设置为接收事件
    60. 		nty_event_del(reactor->epfd, ev);
    61. 		nty_event_set(ev, fd, recv_cb, reactor);
    62. 		nty_event_add(reactor->epfd, EPOLLIN, ev);
    63. 		
    64. 	} else { //发送失败
    65.  
    66. 		nty_event_del(reactor->epfd, ev);
    67. 		close(ev->fd);
    68.  
    69. 		printf("send[fd=%d] error %s\n", fd, strerror(errno));
    70.  
    71. 	}
    72.  
    73. 	return len;
    74. }
    75.  
    76.  
    77. //客户端接入回调
    78. int accept_cb(int fd, int events, void* arg)
    79. {
    80. 	struct ntyreactor *reactor = (struct ntyreactor*)arg;
    81. 	if (reactor == NULL) return -1;
    82.  
    83. 	struct sockaddr_in client_addr;
    84. 	socklen_t len = sizeof(client_addr);
    85.  
    86. 	int clientfd;
    87.  
    88. 	//客户端接入
    89. 	if ((clientfd = accept(fd, (struct sockaddr*)&client_addr, &len)) == -1) {
    90. 		if (errno != EAGAIN && errno != EINTR) {
    91. 			
    92. 		}
    93. 		printf("accept: %s\n", strerror(errno));
    94. 		return -1;
    95. 	}
    96.  
    97. 	//设置非阻塞fd
    98. 	int flag = 0;
    99. 	if ((flag = fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) {
    100. 		printf("%s: fcntl nonblocking failed, %d\n", __func__, MAX_EPOLL_EVENTS);
    101. 		return -1;
    102. 	}
    103.  
    104. 	struct ntyevent *event = ntyreactor_idx(reactor, clientfd);
    105.  
    106. 	if (event == NULL) return -1;
    107.  
    108. 	//将该fd设置为recv
    109. 	nty_event_set(event, clientfd, recv_cb, reactor);
    110. 	nty_event_add(reactor->epfd, EPOLLIN, event);
    111.  
    112. 	printf("new connect [%s:%d], pos[%d]\n", 
    113. 		inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), clientfd);
    114.  
    115. 	return 0;
    116. }

     7.  完整代码和测试结果

    1. #include 
    2. #include 
    3. #include 
    4. #include 
    5. #include 
    6. #include 
    7. #include 
    8. #include 
    9. #include 
    10.  
    11. #define  BUFFER_LENGTH  	1024
    12. #define  MAX_EPOLL_EVENTS   1024  //epoll事件数量
    13. #define  SERVER_PORT  		8888
    14. #define  PORT_COUNT			100
    15.  
    16. typedef int NCALLBACK(int ,int ,void*);
    17.  
    18. //管理每一个io fd的结构体
    19. struct ntyevent{
    20. 	int fd; //io fd
    21. 	int events;
    22. 	void *arg;
    23. 	int (*callback)(int fd, int events, void* arg); //执行回调函数
    24.  
    25. 	int status;	//判断是否已有事件
    26. 	char buffer[BUFFER_LENGTH]; //用户缓冲区
    27. 	int length; //用户缓冲区长度
    28. };
    29.  
    30. //管理ntyevent fd的块
    31. struct eventblock{
    32. 	struct eventblock* next; //指向ntyevent fd集合
    33. 	struct ntyevent* events; //指向下一个ntyevent fd的块
    34. };
    35.  
    36. //reacotr结点
    37. struct ntyreactor{
    38. 	int epfd;	//epoll fd
    39. 	int blkcnt;	//ntyevent fd的块 计数
    40.  
    41. 	struct eventblock* evblks;	//指向ntyevent fd的块头结点
    42. };
    43.  
    44. int recv_cb(int fd, int events, void *arg);
    45. int send_cb(int fd, int events, void *arg);
    46. int accept_cb(int fd, int events, void* arg)；
    47.  
    48. struct ntyevent *ntyreactor_idx(struct ntyreactor *reactor, int sockfd);
    49.  
    50. //io fd结构体设置
    51. void nty_event_set(struct ntyevent* ev, int fd, NCALLBACK callback, void* arg)
    52. {
    53. 	ev->fd = fd;
    54. 	ev->callback = callback;
    55. 	ev->events = 0;
    56. 	ev->arg = arg;
    57.  
    58. 	return ;
    59. }
    60.  
    61. //io fd add
    62. int nty_event_add(int epfd, int events, struct ntyevent *ev)
    63. {
    64. 	struct epoll_event ep_ev = {0, {0}};
    65. 	ep_ev.data.ptr = ev;				//io fd结构体
    66. 	ep_ev.events = ev->events = events; //需要检测的fd事件
    67.  
    68. 	int op; //操作类型
    69. 	if(ev->status == 1){
    70. 		op = EPOLL_CTL_MOD; //修改
    71. 	}else{
    72. 		op = EPOLL_CTL_ADD; //添加
    73. 		ev->status = 1;		//标志已经添加
    74. 	}
    75. 		
    76. 	if(epoll_ctl(epfd, op, ev->fd, &ep_ev) <0 ){ //绑定
    77. 		printf("event add failed [fd=%d], events[%d]\n", ev->fd, events);
    78. 		return -1;
    79. 	}
    80.  
    81. 	return 0;
    82. }
    83.  
    84. //io fd del
    85. int nty_event_del(int epfd, struct ntyevent* ev)
    86. {
    87. 	struct epoll_event ep_ev = {0, {0}};
    88. 	if(ev->status != 1){ //没有添加过检测的fd事件
    89. 		return -1;
    90. 	}
    91.  
    92. 	ep_ev.data.ptr = ev;
    93. 	ev->status = 0; //标志未添加
    94. 	epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, &ep_ev);
    95. 	return 0;
    96. }
    97.  
    98. //recv回调
    99. int recv_cb(int fd, int events, void* arg)
    100. {
    101. 	struct ntyreactor* reactor = (struct ntyreactor*)arg;
    102. 	struct ntyevent* ev = ntyreactor_idx(reactor, fd);	
    103.  
    104. 	if(ev == NULL)return -1;
    105.  
    106. 	int len = recv(fd, ev->buffer, BUFFER_LENGTH, 0);
    107. 	nty_event_del(reactor->epfd, ev);
    108.  
    109. 	if (len > 0) {
    110. 		
    111. 		ev->length = len;
    112. 		ev->buffer[len] = '\0';
    113.  
    114. 		printf("recv [%d]:%s\n", fd, ev->buffer);
    115.  
    116. 		//将fd 设置为发送事件
    117. 		nty_event_set(ev, fd, send_cb, reactor);
    118. 		nty_event_add(reactor->epfd, EPOLLOUT, ev);
    119. 		
    120. 		
    121. 	} else if (len == 0) {  //客户端断开连接
    122.  
    123. 		nty_event_del(reactor->epfd, ev);
    124. 		printf("recv_cb --> disconnect\n");
    125. 		close(ev->fd);
    126. 		 
    127. 	} else { //返回错误
    128.  
    129. 		if (errno == EAGAIN && errno == EWOULDBLOCK) { //
    130. 			
    131. 		} else if (errno == ECONNRESET){
    132. 			nty_event_del(reactor->epfd, ev);
    133. 			close(ev->fd);
    134. 		}
    135. 		printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));
    136. 		
    137. 	}
    138.  
    139. 	return len;
    140. }
    141.  
    142.  
    143.  
    144. //send回调
    145. int send_cb(int fd, int events, void* arg)
    146. {
    147. 	struct ntyreactor* reactor = (struct ntyreactor*)arg;
    148. 	struct ntyevent* ev = ntyreactor_idx(reactor, fd);
    149.  
    150. 	if (ev == NULL) return -1;
    151.  
    152. 	int len = send(fd, ev->buffer, ev->length, 0);
    153. 	if (len > 0) {
    154. 		printf("send[fd=%d], [%d]%s\n", fd, len, ev->buffer);
    155.  
    156. 		//发送后，将fd设置为接收事件
    157. 		nty_event_del(reactor->epfd, ev);
    158. 		nty_event_set(ev, fd, recv_cb, reactor);
    159. 		nty_event_add(reactor->epfd, EPOLLIN, ev);
    160. 		
    161. 	} else { //发送失败
    162.  
    163. 		nty_event_del(reactor->epfd, ev);
    164. 		close(ev->fd);
    165.  
    166. 		printf("send[fd=%d] error %s\n", fd, strerror(errno));
    167.  
    168. 	}
    169.  
    170. 	return len;
    171. }
    172.  
    173.  
    174. //客户端接入回调
    175. int accept_cb(int fd, int events, void* arg)
    176. {
    177. 	struct ntyreactor *reactor = (struct ntyreactor*)arg;
    178. 	if (reactor == NULL) return -1;
    179.  
    180. 	struct sockaddr_in client_addr;
    181. 	socklen_t len = sizeof(client_addr);
    182.  
    183. 	int clientfd;
    184.  
    185. 	//客户端接入
    186. 	if ((clientfd = accept(fd, (struct sockaddr*)&client_addr, &len)) == -1) {
    187. 		if (errno != EAGAIN && errno != EINTR) {
    188. 			
    189. 		}
    190. 		printf("accept: %s\n", strerror(errno));
    191. 		return -1;
    192. 	}
    193.  
    194. 	//设置非阻塞fd
    195. 	int flag = 0;
    196. 	if ((flag = fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) {
    197. 		printf("%s: fcntl nonblocking failed, %d\n", __func__, MAX_EPOLL_EVENTS);
    198. 		return -1;
    199. 	}
    200.  
    201. 	struct ntyevent *event = ntyreactor_idx(reactor, clientfd);
    202.  
    203. 	if (event == NULL) return -1;
    204.  
    205. 	//将该fd设置为recv
    206. 	nty_event_set(event, clientfd, recv_cb, reactor);
    207. 	nty_event_add(reactor->epfd, EPOLLIN, event);
    208.  
    209. 	printf("new connect [%s:%d], pos[%d]\n", 
    210. 		inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), clientfd);
    211.  
    212. 	return 0;
    213. }
    214.  
    215. //创建socket监听
    216. int init_sock(short port)
    217. {
    218. 	int fd = socket(AF_INET, SOCK_STREAM, 0);
    219. 	fcntl(fd, F_SETFL, O_NONBLOCK);
    220.  
    221. 	struct sockaddr_in server_addr;
    222. 	memset(&server_addr, 0, sizeof(server_addr));
    223. 	server_addr.sin_family = AF_INET;
    224. 	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    225. 	server_addr.sin_port = htons(port);
    226.  
    227. 	bind(fd, (struct sockaddr*)&server_addr, sizeof(server_addr));
    228.  
    229. 	if (listen(fd, 20) < 0) {
    230. 		printf("listen failed : %s\n", strerror(errno));
    231. 		return -1;
    232. 	}
    233.  
    234. 	printf("listen server port : %d\n", port);
    235. 	return fd;
    236. }
    237.  
    238.  
    239. //reactor扩展大小
    240. int ntyreactor_alloc(struct ntyreactor* reactor)
    241. {
    242. 	if(reactor == NULL) return -1;
    243. 	if(reactor->evblks == NULL) return -1;
    244.  
    245. 	struct eventblock* blk = reactor->evblks; //块的头结点
    246.  
    247. 	//找尾节点
    248. 	while(blk->next != NULL){  //找到尾节点
    249. 		blk = blk->next;
    250. 	}
    251.  
    252. 	struct ntyevent* evs = (struct ntyevent*)malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    253. 	if (evs == NULL) {
    254. 		printf("ntyreactor_alloc ntyevent failed\n");
    255. 		return -2;
    256. 	}
    257. 	memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    258.  
    259. 	struct eventblock *block = malloc(sizeof(struct eventblock));
    260. 	if (block == NULL) {
    261. 		printf("ntyreactor_alloc eventblock failed\n");
    262. 		return -3;
    263. 	}
    264. 	//io fd集合连接成块
    265. 	block->events = evs;
    266. 	block->next = NULL;
    267.  
    268. 	//指向新块
    269. 	blk->next = block;
    270. 	reactor->blkcnt ++;
    271.  
    272. 	return 0;
    273. }
    274.  
    275. //根据io fd来找fd结构体
    276. struct ntyevent *ntyreactor_idx(struct ntyreactor *reactor, int sockfd) {
    277. 	if (reactor == NULL) return NULL;
    278. 	if (reactor->evblks == NULL) return NULL;
    279.  
    280. 	int blkidx = sockfd / MAX_EPOLL_EVENTS; //在哪一个块
    281. 	while (blkidx >= reactor->blkcnt) {		//大小不够扩容
    282. 		ntyreactor_alloc(reactor);
    283. 	}
    284.  
    285. 	int i = 0;
    286. 	struct eventblock *blk = reactor->evblks; //头结点块
    287. 	while (i++ != blkidx && blk != NULL) {    //找到所在的块
    288. 		blk = blk->next;
    289. 	}
    290.  
    291. 	return &blk->events[sockfd % MAX_EPOLL_EVENTS]; //返回fd结构体
    292. }
    293.  
    294. //reactor初始化
    295. int ntyreactor_init(struct ntyreactor* reactor)
    296. {
    297. 	if(reactor == NULL) return -1;
    298. 	memset(reactor, 0, sizeof(struct ntyreactor));
    299.  
    300. 	reactor->epfd = epoll_create(1); 
    301. 	if (reactor->epfd <= 0) {
    302. 		printf("create epfd in %s err %s\n", __func__, strerror(errno));
    303. 		return -2;
    304. 	}
    305.  
    306. 	//创建第一个块
    307. 	struct ntyevent* evs = (struct ntyevent*)malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    308. 	if (evs == NULL) {
    309. 		printf("create epfd in %s err %s\n", __func__, strerror(errno));
    310. 		close(reactor->epfd);
    311. 		return -3;
    312. 	}
    313. 	memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    314.  
    315. 	struct eventblock *block = malloc(sizeof(struct eventblock));
    316. 	if (block == NULL) {
    317. 		free(evs);
    318. 		close(reactor->epfd);
    319. 		return -3;
    320. 	}
    321. 	block->events = evs;
    322. 	block->next = NULL;
    323.  
    324. 	reactor->evblks = block;
    325. 	reactor->blkcnt = 1;
    326.  
    327. 	return 0;
    328. }
    329.  
    330. //销毁reactor
    331. int ntyreactor_destory(struct ntyreactor* reactor)
    332. {
    333. 	close(reactor->epfd);
    334.  
    335. 	struct eventblock *blk = reactor->evblks;
    336. 	struct eventblock *blk_next;
    337. 	while (blk != NULL) {
    338. 		blk_next = blk->next;
    339.  
    340. 		free(blk->events);
    341. 		free(blk);
    342. 		
    343. 		blk = blk_next;
    344. 	}
    345.  
    346. 	return 0;
    347. }
    348.  
    349. //初始化接收连接socket
    350. int ntyreactor_addlistener(struct ntyreactor* reactor, int sockfd, NCALLBACK *acceptor){
    351. 	if (reactor == NULL) return -1;
    352. 	if (reactor->evblks == NULL) return -1;
    353.  
    354. 	struct ntyevent* event = ntyreactor_idx(reactor, sockfd);
    355. 	if (event == NULL) return -1;
    356.  
    357. 	nty_event_set(event, sockfd, acceptor, reactor);
    358. 	nty_event_add(reactor->epfd, EPOLLIN, event);
    359.  
    360. 	return 0;
    361. }
    362.  
    363. //reactor事件循环
    364. int ntyreactor_run(struct ntyreactor* reactor)
    365. {
    366. 	if (reactor == NULL) return -1;
    367. 	if (reactor->epfd < 0) return -1;
    368. 	if (reactor->evblks == NULL) return -1;
    369. 	
    370. 	struct epoll_event events[MAX_EPOLL_EVENTS+1];
    371. 	
    372. 	int checkpos = 0, i;
    373.  
    374. 	while(1){
    375. 		int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
    376. 		if (nready < 0) {
    377. 			printf("epoll_wait error, exit\n");
    378. 			continue;
    379. 		}
    380.  
    381. 		for(i = 0;i < nready; i++){
    382. 			struct ntyevent* ev = (struct ntyevent*)events[i].data.ptr; //发生事件的io fd结构体
    383.  
    384. 			if((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)){
    385. 				ev->callback(ev->fd, events[i].events, ev->arg);
    386. 			}
    387. 			if((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)){
    388. 				ev->callback(ev->fd, events[i].events, ev->arg);
    389. 			}
    390. 		
    391. 		}
    392.  
    393. 	}
    394.  
    395. }
    396.  
    397. int main(int argc, char *argv[]) {
    398.  
    399. 	struct ntyreactor *reactor = (struct ntyreactor*)malloc(sizeof(struct ntyreactor));
    400. 	ntyreactor_init(reactor);
    401.  
    402. 	//起始的端口号
    403. 	unsigned short port = SERVER_PORT;
    404. 	if (argc == 2) {
    405. 		port = atoi(argv[1]);
    406. 	}
    407.  
    408. 	int i = 0;
    409. 	int sockfds[PORT_COUNT] = {0};
    410.  
    411. 	for (i = 0;i < PORT_COUNT;i ++) {
    412. 		sockfds[i] = init_sock(port+i);
    413. 		ntyreactor_addlistener(reactor, sockfds[i], accept_cb);
    414. 	}
    415.  
    416. 	ntyreactor_run(reactor);
    417.  
    418. 	ntyreactor_destory(reactor);
    419. 	
    420. 	for (i = 0;i < PORT_COUNT;i ++) {
    421. 		close(sockfds[i]);
    422. 	}
    423. 	free(reactor);
    424.  
    425. 	return 0;
    426. }
    427.  
    428.  
    429.  
    430.  
    431.  
    432.

     运行结果：

     

    总结

    本章实现了一个网络经典模型，设计模式reactor 事件循环，事件驱动的反应堆模式.

    组件:

    事件处理器：回调函数callback    

    事件分发器：(将事件分发给对应的事件处理器),

    多路复用器：(select poll epoll 等操作系统提供的多路复用技术)

    流程：

    1. 注册读就绪事件和相应的事件处理器；
    2. 事件分离器等待事件；
    3. 事件到来，激活分离器，分离器调用事件对应的处理器；
    4. 事件处理器完成实际的读操作，处理读到的数据，注册新的事件，然后返还控制
    权。 

  • 相关阅读:
    MAYA教程之建模基础命令介绍
    Bootstrap中的栅格系统
    TiDB 6.0：让 TSO 更高效丨TiDB Book Rush
    JAVA计算机毕业设计毕业生交流学习平台Mybatis+源码+数据库+lw文档+系统+调试部署
    把 Maven 提交到项目？Maven Wrapper的使用与好处
    [项目管理-30]：项目成员成熟度以及采取的不同的策略
    linux查看系统/内核版本号、CPU核数/线程/型号、内存大小等
    成绩统计-蓝桥杯
    ES6中新增加的Map和Set数据结构的使用场景
    微软智能云在华发布多项混合云服务及功能更新
  • 原文地址：https://blog.csdn.net/kakaka666/article/details/126216142