Chinaunix首页 | 论坛 | 博客
  • 博客访问: 276351
  • 博文数量: 95
  • 博客积分: 618
  • 博客等级: 中士
  • 技术积分: 455
  • 用 户 组: 普通用户
  • 注册时间: 2011-03-28 13:39
文章分类

全部博文(95)

文章存档

2015年(65)

2013年(1)

2012年(10)

2011年(19)

分类: C/C++

2015-08-30 00:23:33

原文地址:使用libevent编写Linux服务 作者:txgc_wm

一、初始化事件
  

       首先完成对libenvent的事件初始化和事件驱动模型的选择。在使用多线程的情况下,一般我们需获取所返回的事件根基。

main_base = event_init();

event_init函数返回的是一个event_base对象,该对象包括了事件处理过程中的一些全局变量,其结构为:

struct event_base {
    
const struct eventop *evsel;
    
void *evbase;
    
int event_count;        /* counts number of total events */
    
int event_count_active; /* counts number of active events */
    int event_gotterm;      /* Set to terminate loop */
    
int event_break;        /* Set to terminate loop immediately */
    
/* active event management */
    
struct event_list **activequeues;
    
int nactivequeues;
    
/* signal handling info */
    
struct evsignal_info sig;
    
struct event_list eventqueue;
    
struct timeval event_tv;
    
struct min_heap timeheap;
    
struct timeval tv_cache;
};



二、 添加事件

       事件初始化完毕后,可以使用event_set设置事件,然后使用event_add将其加入。首先完成socket的监听,然后将其加入到事件队列中(这里对所有的异常都不做考虑)。

1socket监听

struct sockaddr_in listen_addr;

int port = 10000//socket监听端口
int listen_fd = socket(AF_INET, SOCK_STREAM, 0);

memset(
&listen_addr, 0sizeof(listen_addr));

listen_addr.sin_family 
= AF_INET;
listen_addr.sin_addr.s_addr 
= INADDR_ANY;
listen_addr.sin_port 
= htons(port)

reuseaddr_on 
= 1;

setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, 
&reuseaddr_on, sizeof(reuseaddr_on)) #支持端口复用

bind(listen_fd, (
struct sockaddr *&listen_addr, sizeof(listen_addr))'
listen(listen_fd, 
1024);

/*将描述符设置为非阻塞*/
int flags = fcntl(listen_fd,F_GETFL);
flags 
|= O_NONBLOCK;
fcntl(listen_fd, F_SETFL, flags);

2)事件设置

       socket服务建立后,就可以进行事件设置。使用event_set来设置事件对象,其传入参数包括事件根基(event_base对象),描述符,事件类型,事件发生时的回调函数,回调函数传入参数。其中事件类型包括EV_READEV_WRITEEV_PERSISTEV_PERSIST和前两者结合使用,表示该事件为持续事件。

struct event ev;

event_set(&ev, listen_fd, EV_READ | EV_PERSIST, accept_handle, (void *)&ev);

3)事件添加与删除

       事件设置好后,就可以将其加入事件队列。event_add用来将事件加入,它接受两个参数:要添加的事件和时间的超时值。 如果需要将事件删除,可以使用event_del来完成。event_del函数会取消所指定的事件。

event_add(&ev, NULL)



三、进入事件循环

     事件成功添加后就是万事具备只欠东风了,libevent提供了多种方式来进入事件循环,常用的是event_dispatchevent_base_loop,前者最后实际是使用当前事件根基来调用event_base_loop

   event_base_loop(main_base, 0);



四、处理连接

       到这里为止,已经完成了事件的设置、事件的添加并进入到了事件循环。但是当事件发生时(这里就是连接建立)如何处理呢? 聪明的用户会想到前面我们在事件设置时指定的回调函数accept_handle。没错,当连接建立时回调函数accept_handle会自动的得到调用。

        对于缓冲区的读写在非阻塞式网络编程中是一个难以处理的问题,幸运的是libevent提供了buffereventevbuf来替我们完成该项工作。这里我们采用bufferevent来处理。

(1)生成bufferevent对象

      使用bufferevent_new对象来生成bufferevent对象,并分别指定读、写、连接错误时的处理函数和函数传入参数。

(2) 设置读取量

       bufferevent的读事件激活以后,即使用户没有读取完bufferevent缓冲区中的数据, bufferevent读事件也不会再次被激活。因为bufferevent的读事件是由其所监控的描述符的读事件激活的,只有描述符可读,读事件才会被激活。可通过设置wm_read.high来控制bufferevent从描述符缓冲区中读取的数据量。

(3) 将事件加入事件队列

       和前面一样,在事件设置好后,需将事件加入到事件队列中, 不过bufferevent的有自己专门的加入函数bufferevent_base_set和激活函数bufferevent_enable

       bufferevent接收两个参数事件根基个事件对象,前者用来指定事件将加入到哪个事件根基中,后者说明需将那个bufferevnet事件加入。(在多线程的情况下,每个线程可能有自己单独的事件根基)

       在bufferevent初始化完毕后,可以使用bufferevent_enablebufferevent_disable反复的激活与禁止事件,其接收参数为事件对象和事件标志。其中标志参数为EV_READEV_WRITE

void accept_handle(const int sfd, const short eventvoid *arg)
{
   
struct sockaddr_in addr;

   socklen_t addrlen 
= sizeof(addr);

   
int fd = accept(sfd, (struct sockaddr *&addr, &addrlen); //处理连接

    buf_ev 
= bufferevent_new(fd,   buffered_on_read, NULL, NULL, fd)
    buf_ev
->wm_read.high = 4096
    bufferevent_base_set(main_base, buf_ev);
    bufferevent_enable(buf_ev, EV_READ);
}



五、 读取缓冲区

       当缓冲区读就绪时会自动激活前面注册的缓冲区读函数,我们可以使用bufferevent_read函数来读取缓冲区bufferevent_read函数参数分别为:所需读取的事件缓冲区,读入数据的存放地,希望读取的字节数。函数返回实际读取的字节数。注意:及时缓冲区未读完,事件也不会再次被激活(除非再次有数据)。因此此处需反复读取直到全部读取完毕。



六、写回客户端

       bufferevent系列函数不但支持读取缓冲区,而且支持写缓冲区(即将结果返回给客户端)。

void buffered_on_read(struct bufferevent *bev, void * arg){
    
char buffer[4096]

    ret 
= bufferevent_read(bev, &buffer, 4096);
    bufferevent_write(bef, (
void *)&buffer, 4096);
}

       至此已经可以使用libevent编写非阻塞的事件驱动服务器,它支持连接建立、socket可读等事件的处理。但在实际的使用事件驱动的服务器中,通常是使用一个线程处理连接,然后使用多个线程来处理请求。



七、异步事件处理示例

利用libevent编写服务端程序,主要有4部分
a.创建主通知链base
  base = event_base_new();
 
b.创建要监听的事件,并将其加入到主通知链中。
  listener_event = event_new(base, listener, EV_READ|EV_PERSIST, do_accept, (void*)base);
  event_add(listener_event, NULL);
  event_free( listener_event ); //释放由event_new申请的结构体
 
c.主循环
  event_base_dispatch(base);

d.释放
  event_base_free(base);

      以下程序中do_read, do_write是异步的,为了解决了异步之间的问题,程序使用了state这个结构体变量将do_read和do_write联系起来。
  1. #include <netinet/in.h>
  2. #include <sys/socket.h>
  3. #include <event2/event.h>
  4. #include <assert.h>
  5. #include <unistd.h>
  6. #include <string.h>
  7. #include <stdlib.h>
  8. #include <stdio.h>
  9. #include <errno.h>
  10. #include <fcntl.h>

  11. #define MAX_LINE 16384
  12. #define PORT 9999

  13. void do_read(evutil_socket_t fd, short events, void *arg);
  14. void do_write(evutil_socket_t fd, short events, void *arg);

  15. struct fd_state {
  16.     char buffer[MAX_LINE];
  17.     size_t buffer_used;

  18.     size_t n_written;
  19.     size_t write_upto;

  20.     struct event *read_event;
  21.     struct event *write_event;
  22. };

  23. struct fd_state *alloc_fd_state(struct event_base *base, evutil_socket_t fd)
  24. {
  25.     struct fd_state *state =
  26.      (struct fd_state *)malloc(sizeof(struct fd_state));
  27.     if (!state) {
  28.         return NULL;
  29.     }

  30.     state->read_event =
  31.      event_new(base, fd, EV_READ | EV_PERSIST, do_read, state);
  32.     if (!state->read_event) {
  33.         free(state);
  34.         return NULL;
  35.     }

  36.     state->write_event =
  37.      event_new(base, fd, EV_WRITE | EV_PERSIST, do_write, state);
  38.     if (!state->write_event) {
  39.         event_free(state->read_event);
  40.         free(state);
  41.         return NULL;
  42.     }

  43.     assert(state->write_event);

  44.     return state;
  45. }

  46. void free_fd_state(struct fd_state *state)
  47. {
  48.     event_free(state->read_event);
  49.     event_free(state->write_event);
  50.     free(state);
  51. }

  52. void do_read(evutil_socket_t fd, short events, void *arg)
  53. {
  54.     struct fd_state *state = arg;
  55.     char buf[1024];
  56.     int i;
  57.     ssize_t result;
  58.     while (1) {
  59.         // assert(state->write_event);
  60.         result = recv(fd, buf, sizeof(buf), 0);
  61.         if (result <= 0)
  62.             break;
  63.         printf("[%s][%d]buf=[%s]len=[%d]\n", __FILE__, __LINE__, buf,
  64.          result);
  65.     }

  66.     memcpy(state->buffer, "reply", sizeof("reply"));
  67.     assert(state->write_event);
  68.     event_add(state->write_event, NULL);
  69.     state->write_upto = state->buffer_used;

  70.     if (result == 0) {
  71.         free_fd_state(state);
  72.     } else if (result < 0) {
  73.         if (errno == EAGAIN)    // XXXX use evutil macro
  74.             return;
  75.         perror("recv");
  76.         free_fd_state(state);
  77.     }
  78. }

  79. void do_write(evutil_socket_t fd, short events, void *arg)
  80. {
  81.     struct fd_state *state = arg;

  82.     printf("[%d]\n", __LINE__);
  83.     //while (state->n_written < state->write_upto)
  84.     {
  85.         //ssize_t result = send(fd, state->buffer + state->n_written,
  86.         //state->write_upto - state->n_written, 0);
  87.         ssize_t result =
  88.          send(fd, state->buffer, strlen(state->buffer), 0);
  89.         if (result < 0) {
  90.             if (errno == EAGAIN)    // XXX use evutil macro
  91.                 return;
  92.             free_fd_state(state);
  93.             return;
  94.         }
  95.         assert(result != 0);
  96.         state->n_written += result;
  97.     }

  98.     //if (state->n_written == state->buffer_used)
  99.     {
  100.         state->n_written = state->write_upto = state->buffer_used = 1;
  101.     }

  102.     event_del(state->write_event);
  103. }

  104. void do_accept(evutil_socket_t listener, short event, void *arg)
  105. {
  106.     struct event_base *base = arg;
  107.     struct sockaddr_storage ss;
  108.     socklen_t slen = sizeof(ss);

  109.     int fd = accept(listener, (struct sockaddr *)&ss, &slen);
  110.     if (fd < 0) {        // XXXX eagain??
  111.         perror("accept");
  112.     } else if (fd > FD_SETSIZE) {
  113.         close(fd);    // XXX replace all closes with EVUTIL_CLOSESOCKET */
  114.     } else {
  115.         struct fd_state *state;
  116.         evutil_make_socket_nonblocking(fd);
  117.         state = alloc_fd_state(base, fd);

  118.         assert(state);    /*XXX err */
  119.         assert(state->write_event);
  120.         event_add(state->read_event, NULL);
  121.     }
  122. }

  123. void run(void)
  124. {
  125.     evutil_socket_t listener;
  126.     struct sockaddr_in sin;
  127.     struct event_base *base;
  128.     struct event *listener_event;

  129.     base = event_base_new();
  130.     if (!base)
  131.         return;        /*XXXerr */

  132.     sin.sin_family = AF_INET;
  133.     sin.sin_addr.s_addr = 0;
  134.     sin.sin_port = htons(PORT);

  135.     listener = socket(AF_INET, SOCK_STREAM, 0);
  136.     evutil_make_socket_nonblocking(listener);

  137. #ifndef WIN32
  138.     {
  139.         int one = 1;
  140.         setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &one,
  141.              sizeof(one));
  142.     }
  143. #endif

  144.     if (bind(listener, (struct sockaddr *)&sin, sizeof(sin)) < 0) {
  145.         perror("bind");
  146.         return;
  147.     }

  148.     if (listen(listener, 16) < 0) {
  149.         perror("listen");
  150.         return;
  151.     }

  152.     listener_event =
  153.      event_new(base, listener, EV_READ | EV_PERSIST, do_accept,
  154.          (void *)base);
  155.     /*XXX check it */
  156.     event_add(listener_event, NULL);
  157.     event_base_dispatch(base);
  158.     event_base_free(base);

  159.     return;
  160. }

  161. int main(int argc, char **argv)
  162. {
  163.     setvbuf(stdout, NULL, _IONBF, 0);
  164.     run();

  165.     return 0;
  166. }
阅读(1495) | 评论(0) | 转发(0) |
0

上一篇:libevent介绍及示例

下一篇:没有了

给主人留下些什么吧!~~