Memcached的服务设计与启动过程——C10K系列

luohnoyo 2014-08-23

C10k要解决的问题,是10K个连接。

LINUX下,使用EPOLL可实现异步非阻塞(注:阻塞的一定是同步的,阻塞是调用方自己阻塞自己(等待事件))

非阻塞:是指调用方不会阻塞自己,如被调用方有数据就返回,无数据就返回EAGAIN,调用方根据EAGAIN决定自己的策略。因此非阻塞,和异步没有任何关联。

异步:是相对于同步的。异步是指:调用的时机和返回的时机不是同一时刻。异步说的是一个处理流程,而并不一定是具体的某个函数。

举例如下:

你在电子商城买了一本书,你已经下单;但是,这时候商城并没有立即给你发单,而是有合适时候时,主动通知你,我已经发单了。

因此,下单,和发单不是同一时刻。

一般来说,异步需要通过回调函数来实现。即先需要注册一个异步处理流程,如必须先将你的手机号告诉给商城。

当然,也有异步服务。比如用网银买火车票。火车票网站转到网银,网银付账是一个同步的过程。但是火车票网站与网银的处理是异步的。火车票网站,不知道网银交易何时完成;网银交易完成后,再发个消息给火车票网站交易完成。

---------------------------------------------------------------------------------

Memcached,版本1.4.20:基于libevent的多线程实现

采用的是一个主线程(dispatcher_thread),多个工作线程worker(LIBEVENT_THREAD),

(1)主线程dispatcher_thread负责接收外部请求事件,然后派发给工作线程处理。

(2)主线程dispatcher_thread和工作线程的通过管道传递消息。(实际上,是工作线程添加了一个与主线程关联的管道的事件(libevent的event))

对于TCP连接来说,主线程(dispatcher_thread)是一个监听线程,即监听外部的服务请求。

主线程(dispatcher_thread)的定义:

typedefstruct{

pthread_tthread_id;

/*uniqueIDofthisthread*/

structevent_base*base;

/*libeventhandlethisthreaduses*/

}LIBEVENT_DISPATCHER_THREAD;

工作线程Worker的定义:

typedefstruct{

pthread_tthread_id;

/*uniqueIDofthisthread*/

structevent_base*base;

/*libeventhandlethisthreaduses*/

structeventnotify_event;/*listeneventfornotifypipe*/

intnotify_receive_fd;

/*receivingendofnotifypipe*/

intnotify_send_fd;

/*sendingendofnotifypipe*/

structconn_queue*new_conn_queue;/*queueofnewconnectionstohandle*/

}LIBEVENT_THREAD;

Libevent的main函数比较长,但与网络请求相关的是serversocket的创建和thread_init函数。

第一步:多线程服务初始化thread_init()

服务的初始化是在thread_init()完成的,主要完成主线程(dispatcher_thread),多个工作线程worker(LIBEVENT_THREAD)事件注册,管道通知等:

(1)初始化主线程dispatcher_thread的eventbase,event;

(2)初始化管道,将主线程dispatcher_thread与工作线程worker,通过管道关联。

for(i=0;inotify_event,me->notify_receive_fd,

EV_READ|EV_PERSIST,thread_libevent_process,me);

event_base_set(me->base,&me->notify_event);

if(event_add(&me->notify_event,0)==-1){

fprintf(stderr,"Can'tmonitorlibeventnotifypipe\n");

exit(1);

}

§thread_libevent_process:这个函数,是接收到数据的具体处理过程,其参数是线程句柄

§初始化工作线程自己的请求队列。

LIBEVENT_THREAD中的structconn_queue*new_conn_queue;这个是请求队列。

(4)启动工作线程

创建Worker线程(LIBEVENT_THREAD),其函数入口是:

staticvoid*worker_libevent(void*arg)

里面是一个event_base_loop()的消息循环。

(5)到这里,工作线程已经完成初始化和启动(即工作线程已经可以工作了)。主线程只要等待工作线程启动完毕。

第二步,serversocket创建:和所有网络服务一样:createthelisteningsocket,bindit,andinit

server_sockets(settings.port,tcp_transport,

portnumber_file)

new_socket:

获取socket的设置:fcntl(sfd,F_GETFL,0))

将其设置为非阻塞:fcntl(sfd,F_SETFL,flags|O_NONBLOCK)

socket设置:根据配置文件

SO_REUSEADDR,//地址重用

SO_KEEPALIVE,//是否保持连接,这是TCP层的控制

SO_LINGER,//延时断开连接。ngingx默认也启用

TCP_NODELAY,//不启用Neagle算法(本科书中的滑动窗口拥塞控制协议),一般内网访问,常设置为true

bind:

listen:

listen_conn_add=conn_new(sfd,conn_listening,

EV_READ|EV_PERSIST,1,

transport,main_base);//main_base要和创建的serversocket关联

这里的

conn_new函数:

初始化conn(一个超级复杂的结构体)

设置事件与socket的fd,与main_base关联,设置事件响应的具体处理函数,event_handler

添加事件

到这里,才把main_base(event_base*类型)与监听事件关联完成。

第三步:

主消息循环;整个服务启动过程结束。

备注:主线程dispatcher_thread未必是程序启动的mainthread,可以是一个独立的thread。

相关推荐