Oracle 系统管理 - Linux 系统 - Backtrack 5 - 安全 - Juniper 技术 - Cisco 技术 - 思科模拟器 - Cisco 认证 - Cisco ios 下载

您现在的位置是:Docker > Jenkins > redis ae事件分析

redis ae事件分析

时间:2018-08-06 10:29  来源:未知  阅读次数: 复制分享 我要评论

0. 前言

Redis在封装事件的处理采用了Reactor模式,添加了定时事件的处理。Redis处理事件是单进程单线程的,而经典Reator模式对事件是串行处理的。即如果有一个事件阻塞过久的话会导致整个Redis被阻塞。

对于Reactor模式,可以查看维基百科。(这里推荐一本书:《面向模式的软件架构:卷2》,里面详细讲了Reactor模式,但是刚开始可能会不太 明白,耐下心来就好)。我自己也使用Reactor封装了一个事件处理,其中对于读事件加入了线程池的处理,但是还在测试(该封装主要针对公司需求来 的)。加入线程池的处理是参考了Nettty的事件处理,其也是Reactor+线程池。

补充一下,对于Redis的源码分析:《Redis设计与实现》讲得挺好的。

涉及的文件:ae.h/ae.c ae_epoll.c

涉及的知识点:epoll,Reactor

完成ae事件代码注释在 GitHud 上

1. 数据结构

对于文件和时间的事件结构体,其中的包含了一些函数指针,当对应的事件产生的时候会调用该函数(回调机制)。对于aeEventLoop结构体,比较重 要的几个字段是events,fired,timeEventHead。 对于events,起的是分离表的作用,即通过fd直接映射找到对应的 IOEvent事件结构体(这里直接索引是因为在Linux/Unix类系统中fd是连续的,可以直接作为数组下标)。

fired保存了由底层IO多路复用返回的触发事件(这里我只分析了用epoll的实现)。

timeEventHead保存了定时的事件链表,Redis只用到了2个定时事件。


	
  1. 1 //前置声明,避免了编译出错,因为aeEventLoop需要用到aeFileEvent结构体
  2. 2 struct aeEventLoop;
  3. 3
  4. 4 /* Types and data structures */
  5. 5 //定义文件事件处理接口(函数指针)
  6. 6 typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
  7. 7 //时间事件处理接口(函数指针),该函数返回定时的时长
  8. 8 typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
  9. 9 typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientData);
  10. 10 //aeMain中使用,在调用处理事件前调用
  11. 11 typedef void aeBeforeSleepProc(struct aeEventLoop *eventLoop);
  12. 12
  13. 13 /* File event structure */
  14. 14 //文件事件结构体
  15. 15 typedef struct aeFileEvent {
  16. 16 //读或者写,也用于标识该事件结构体是否正在使用
  17. 17 int mask; /* one of AE_(READABLE|WRITABLE) */
  18. 18 //读事件的处理函数
  19. 19 aeFileProc *rfileProc;
  20. 20 //写事件的处理函数
  21. 21 aeFileProc *wfileProc;
  22. 22 //传递给上述两个函数的数据
  23. 23 void *clientData;
  24. 24 } aeFileEvent;
  25. 25
  26. 26 /* Time event structure */
  27. 27 //时间事件
  28. 28 typedef struct aeTimeEvent {
  29. 29 //时间事件标识符,用于唯一标识该时间事件,并且用于删除时间事件
  30. 30 long long id; /* time event identifier. */
  31. 31 long when_sec; /* seconds */
  32. 32 long when_ms; /* milliseconds */
  33. 33 //该事件对应的处理程序
  34. 34 aeTimeProc *timeProc;
  35. 35 //时间事件的最后一次处理程序,若已设置,则删除时间事件时会被调用
  36. 36 aeEventFinalizerProc *finalizerProc;
  37. 37 void *clientData;
  38. 38 struct aeTimeEvent *next;
  39. 39 } aeTimeEvent;
  40. 40
  41. 41 /* A fired event */
  42. 42 //这里用于保存已触发的事件
  43. 43 typedef struct aeFiredEvent {
  44. 44 int fd;
  45. 45 int mask;
  46. 46 } aeFiredEvent;
  47. 47
  48. 48 /* State of an event based program */
  49. 49 typedef struct aeEventLoop {
  50. 50 //最大文件描述符的值
  51. 51 int maxfd; /* highest file descriptor currently registered */
  52. 52 //文件描述符的最大监听数
  53. 53 int setsize; /* max number of file descriptors tracked */
  54. 54 //用于生成时间事件的唯一标识id
  55. 55 long long timeEventNextId;
  56. 56 //用于检测系统时间是否变更(判断标准 now<lastTime)
  57. 57 time_t lastTime; /* Used to detect system clock skew */
  58. 58 //注册要使用的文件事件,这里的分离表实现为直接索引,即通过fd来访问,实现事件的分离
  59. 59 aeFileEvent *events; /* Registered events */
  60. 60 //已触发的事件
  61. 61 aeFiredEvent *fired; /* Fired events */
  62. 62 aeTimeEvent *timeEventHead;
  63. 63 //停止标志,1表示停止
  64. 64 int stop;
  65. 65 //这个是处理底层特定API的数据,对于epoll来说,该结构体包含了epoll fd和epoll_event
  66. 66 void *apidata; /* This is used for polling API specific data */
  67. 67 //在调用processEvent前(即如果没有事件则睡眠),调用该处理函数
  68. 68 aeBeforeSleepProc *beforesleep;
  69. 69 } aeEventLoop;

2. API实现

a. aeCreateEventLoop

底层epoll多路复用初始化,然后存放在aeEventLoop中 void * 类型的apidata,隐藏了底层的实现。


	
  1. 1 typedef struct aeApiState {
  2. 2 int epfd;
  3. 3 struct epoll_event *events;
  4. 4 } aeApiState;
  5. 5
  6. 6 //ae底层的数据创建以及初始化
  7. 7 static int aeApiCreate(aeEventLoop *eventLoop) {
  8. 8 aeApiState *state = zmalloc(sizeof(aeApiState));
  9. 9
  10. 10 if (!state) return -1;
  11. 11 //创建setsize个epoll_event
  12. 12 state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);
  13. 13 if (!state->events) {
  14. 14 zfree(state);
  15. 15 return -1;
  16. 16 }
  17. 17 state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */
  18. 18 if (state->epfd == -1) {
  19. 19 zfree(state->events);
  20. 20 zfree(state);
  21. 21 return -1;
  22. 22 }
  23. 23 eventLoop->apidata = state;
  24. 24 return 0;
  25. 25 }

	
  1. 1 //创建事件循环,setsize为最大事件的的个数,对于epoll来说也是epoll_event的个数
  2. 2 aeEventLoop *aeCreateEventLoop(int setsize) {
  3. 3 aeEventLoop *eventLoop;
  4. 4 int i;
  5. 5
  6. 6 //分配该结构体的内存空间
  7. 7 if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;
  8. 8 eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);
  9. 9 eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
  10. 10 if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;
  11. 11
  12. 12 //初始化
  13. 13 eventLoop->setsize = setsize; //最多setsize个事件
  14. 14 eventLoop->lastTime = time(NULL);
  15. 15 eventLoop->timeEventHead = NULL;
  16. 16 eventLoop->timeEventNextId = 0;
  17. 17 eventLoop->stop = 0;
  18. 18 eventLoop->maxfd = -1;
  19. 19 eventLoop->beforesleep = NULL;
  20. 20
  21. 21 //这一步为创建底层IO处理的数据,如epoll,创建epoll_event,和epfd
  22. 22 if (aeApiCreate(eventLoop) == -1) goto err;
  23. 23 /* Events with mask == AE_NONE are not set. So let's initialize the
  24. 24 * vector with it. */
  25. 25 for (i = 0; i < setsize; i++)
  26. 26 eventLoop->events[i].mask = AE_NONE;
  27. 27 return eventLoop;
  28. 28
  29. 29 err:
  30. 30 if (eventLoop) {
  31. 31 zfree(eventLoop->events);
  32. 32 zfree(eventLoop->fired);
  33. 33 zfree(eventLoop);
  34. 34 }
  35. 35 return NULL;
  36. 36 }

b. aeCreateFileEvent

对于创建文件事件,需要传入一个该事件对应的处理程序,当事件发生时,会调用对应的回调函数。这里设计的aeFileEvent结构体就是将事件源(FD),事件,事件处理程序关联起来。

 1 //添加监听的事件,其中如果该fd对应的事件已经存在,则为修改合并旧的事件
 2 static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
 3     aeApiState *state = eventLoop->apidata;
 4     struct epoll_event ee;
 5     /* If the fd was already monitored for some event, we need a MOD
 6      * operation. Otherwise we need an ADD operation. */
 7     //判断fd是否已经添加了事件的监听
 8     int op = eventLoop->events[fd].mask == AE_NONE ?
 9             EPOLL_CTL_ADD : EPOLL_CTL_MOD;
10 
11     ee.events = 0;
12     mask |= eventLoop->events[fd].mask; /* Merge old events */
13     if (mask & AE_READABLE) ee.events |= EPOLLIN;
14     if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
15     ee.data.u64 = 0; /* avoid valgrind warning */
16     ee.data.fd = fd;
17     if (epoll_ctl(state->epfd,op,fd,ⅇ) == -1) return -1;
18     return 0;
19 }
20 
21 //删除指定事件的监听
22 static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) {
23     aeApiState *state = eventLoop->apidata;
24     struct epoll_event ee;
25     int mask = eventLoop->events[fd].mask & (~delmask);
26 
27     ee.events = 0;
28     if (mask & AE_READABLE) ee.events |= EPOLLIN;
29     if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
30     ee.data.u64 = 0; /* avoid valgrind warning */
31     ee.data.fd = fd;
32     if (mask != AE_NONE) {
33         epoll_ctl(state->epfd,EPOLL_CTL_MOD,fd,&ee);
34     } else {
35         /* Note, Kernel < 2.6.9 requires a non null event pointer even for
36          * EPOLL_CTL_DEL. */
37         epoll_ctl(state->epfd,EPOLL_CTL_DEL,fd,&ee);
38     }
39 }
 1 //创建文件事件,并将该事件注册到eventLoop中
 2 int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
 3         aeFileProc *proc, void *clientData)
 4 {
 5     if (fd >= eventLoop->setsize) {
 6         errno = ERANGE;
 7         return AE_ERR;
 8     }
 9     //直接使用fd来获取FileEvent,来后面分离事件时也采用这种方法(直接索引)
10     aeFileEvent *fe = &eventLoop->events[fd];
11 
12     //该该事件添加eventLoop中或者修改原来的已有的(保留旧的)
13     if (aeApiAddEvent(eventLoop, fd, mask) == -1)
14         return AE_ERR;
15 
16     fe->mask |= mask;
17     //将该事件的处理程序放到对应的位置
18     if (mask & AE_READABLE) fe->rfileProc = proc;
19     if (mask & AE_WRITABLE) fe->wfileProc = proc;
20     //设置将要传递给该事件处理程序的数据
21     fe->clientData = clientData;
22 
23     if (fd > eventLoop->maxfd)
24         eventLoop->maxfd = fd;
25     return AE_OK;
26 }

c. aeProcessEventsz

这个是核心部分,通过epoll_wait将事件分离出来,从而保存到fired中,对于语句 aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd]; 通过触发事件的 fd 在events中直接映射找到与事件关联的结构体,从而实现事件分派。Reactor的核心是实现了事件的分离分派。

如下流程图:

 1 static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
 2     aeApiState *state = eventLoop->apidata;
 3     int retval, numevents = 0;
 4 
 5     //等待事件产生
 6     retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
 7             tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
 8     if (retval > 0) {
 9         int j;
10 
11         numevents = retval;
12         for (j = 0; j < numevents; j++) {
13             int mask = 0;
14             struct epoll_event *e = state->events+j;
15 
16             if (e->events & EPOLLIN) mask |= AE_READABLE;
17             if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
18             if (e->events & EPOLLERR) mask |= AE_WRITABLE;
19             if (e->events & EPOLLHUP) mask |= AE_WRITABLE;
20             eventLoop->fired[j].fd = e->data.fd;
21             eventLoop->fired[j].mask = mask;
22         }
23     }
24     return numevents;
25 }
 1  //事件处理程序
 2 int aeProcessEvents(aeEventLoop *eventLoop, int flags)
 3 {
 4     int processed = 0, numevents;
 5 
 6     /* Nothing to do? return ASAP */
 7     //若什么都没有设置,则直接返回
 8     if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
 9 
10     /* Note that we want call select() even if there are no
11      * file events to process as long as we want to process time
12      * events, in order to sleep until the next time event is ready
13      * to fire. */
14     if (eventLoop->maxfd != -1 ||
15         ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
16         //如果有文件事件或者设置了时间事件并且没有设置DONT_WAIT标志
17         int j;
18         aeTimeEvent *shortest = NULL;
19         struct timeval tv, *tvp;
20 
21         if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
22             //查找时间最早的时间事件
23             shortest = aeSearchNearestTimer(eventLoop);
24         if (shortest) {
25             long now_sec, now_ms;
26 
27             /* Calculate the time missing for the nearest
28              * timer to fire. */
29             //获取当前时间
30             aeGetTime(&now_sec, &now_ms);
31             tvp = &tv;
32             tvp->tv_sec = shortest->when_sec - now_sec;
33             if (shortest->when_ms < now_ms) {
34                 tvp->tv_usec = ((shortest->when_ms+1000) - now_ms)*1000;
35                 tvp->tv_sec --;
36             } else {
37                 tvp->tv_usec = (shortest->when_ms - now_ms)*1000;
38             }
39 
40             //这里好像有点问题,当sec小于0时表示不用等待,应该将usec也设置为0的
41             //usec比较小,影响不大
42             if (tvp->tv_sec < 0) tvp->tv_sec = 0;
43             if (tvp->tv_usec < 0) tvp->tv_usec = 0;
44 
45         } else {
46             /* If we have to check for events but need to return
47              * ASAP because of AE_DONT_WAIT we need to set the timeout
48              * to zero */
49             if (flags & AE_DONT_WAIT) {
50                 //不等待,直接返回
51                 tv.tv_sec = tv.tv_usec = 0;
52                 tvp = &tv;
53             } else {
54                 //如果没有时间事件则可以阻塞
55                 /* Otherwise we can block */
56                 tvp = NULL; /* wait forever */
57             }
58         }
59 
60         numevents = aeApiPoll(eventLoop, tvp);
61         for (j = 0; j < numevents; j++) {
62             aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
63             int mask = eventLoop->fired[j].mask;
64             int fd = eventLoop->fired[j].fd;
65             int rfired = 0;
66 
67         /* note the fe->mask & mask & ... code: maybe an already processed
68              * event removed an element that fired and we still didn't
69              * processed, so we check if the event is still valid. */
70             if (fe->mask & mask & AE_READABLE) {
71                 rfired = 1;
72                 fe->rfileProc(eventLoop,fd,fe->clientData,mask);
73             }
74             if (fe->mask & mask & AE_WRITABLE) {
75                 if (!rfired || fe->wfileProc != fe->rfileProc)
76                     //这里的判断是为了防止重复调用
77                     fe->wfileProc(eventLoop,fd,fe->clientData,mask);
78             }
79             processed++;
80         }
81     }
82     /* Check time events */
83     if (flags & AE_TIME_EVENTS)
84         processed += processTimeEvents(eventLoop);
85 
86     return processed; /* return the number of processed file/time events */
87 }

3. 总结

1. Reactor模式,串行处理事件

2. 具有定时事件功能(但是不能过多,因为是使用链表实现的)

3. 优先处理读事件

相关资讯