【队列源码研究】消息队列beanstalkd源码详解

顺风车运营研发团队李乐

1.消息队列简介

计算机软件发展的一个重要目标是降低软件耦合性&＃xff1b;

网站架构中&＃xff0c;系统解耦合的重要手段就是异步&＃xff0c;业务之间的消息传递不是同步调用&＃xff0c;而是将一个业务操作分为多个阶段&＃xff0c;每个阶段之间通过共享数据的方式异步执行&＃xff1b;

在分布式系统中&＃xff0c;多个服务器集群通过分布式消息队列实现异步&＃xff1b;分布式消息队列可以看作是内存队列的分布式部署&＃xff1b;

分布式消息队列架构图通常如下所示&＃xff1a;

消息队列是典型的生产者消费者模式&＃xff0c;两者不存在直接调用&＃xff0c;只要保持数据结构不变&＃xff0c;彼此功能实现可以随意改变而不互相影响&＃xff1b;异步消息队列还有以下特点&＃xff1a;

提高系统可用性&＃xff1a;消费者服务器发生故障时&＃xff0c;生产者服务器可以继续处理业务请求&＃xff0c;系统整体表现无故障&＃xff1b;此时数据会在消息队列服务器堆积&＃xff0c;待消费者服务器恢复后&＃xff0c;可以继续处理消息队列中的数据&＃xff1b;
加快网站相应速度&＃xff1a;业务处理前端的生产者服务器在处理完业务请求后&＃xff0c;将数据写入消息队列&＃xff0c;不需要等待消费者服务器处理就可以返回&＃xff0c;减少响应延迟&＃xff1b;
消除并发访问高峰&＃xff1a;用户访问是随机的&＃xff0c;存在高峰和低谷&＃xff1b;可以使用消息队列将突然增加的访问请求数据放入消息队列中&＃xff0c;等待消费者服务器依次处理&＃xff1b;

消费者消费消息时&＃xff0c;通常有两种模式可以选择&＃xff1a;拉模型与推模型。

拉模型是由消息的消费者发起的&＃xff0c;主动权把握在消费者手中&＃xff0c;它会根据自己的情况对生产者发起调用&＃xff1b;
推模式消费者只会被动接受消息&＃xff0c;消息队列一旦发现消息进入&＃xff0c;就会通知消费者执行对消息的处理&＃xff1b;

2.beanstalkd基本知识

2.1beanstalkd简介

beanstalkd是一个轻量级的消息队列&＃xff1b;主要有一下特点&＃xff1a;

拉模式&＃xff0c;消费者需要主动从服务器拉取消息数据&＃xff1b;
tube&＃xff1a;类似于消息主题topic&＃xff0c;一个beanstalkd中可以支持多个tube&＃xff0c;每个tube都有自己的producer和consumer&＃xff1b;多个生产者可以往同一个tube生产job&＃xff0c;多个消费者也能监听同一个tube获取job&＃xff1b;
job&＃xff1a;代替了传统的message&＃xff0c;与消息最大的区别是&＃xff0c;job有多种状态&＃xff1b;
conn&＃xff1a;代表一个客户端链接&＃xff1b;
优先级&＃xff1a;job可以有0~2^32个优先级&＃xff0c;0代表最高优先级&＃xff0c;beanstalkd使用堆处理job的优先级排序&＃xff0c;因此reserve命令的时间复杂度是O(logN)&＃xff1b;
延时&＃xff1a;生产者发布任务时可以指定延时&＃xff0c;到达延迟时间后&＃xff0c;job才能被消费者消费&＃xff1b;
超时机制&＃xff1a;消费者从beanstalkd获取一个job后&＃xff0c;必须在预设的 TTR (time-to-run) 时间内处理完任务&＃xff0c;并发送 delete / release/ bury 命令改变任务状态&＃xff1b;否则 Beanstalkd 会认为消息消费失败&＃xff0c;重置job状态&＃xff0c;使其可以被其他消费者消费。如果消费者预计在 TTR (time-to-run) 时间内无法完成任务, 也可以发送 touch 命令, 它的作用是让 Beanstalkd 从重新计时TTR&＃xff1b;
暂停&＃xff1a;pause命令可以暂停当前tube&＃xff0c;暂停时期内所有job都不能够被消费者消费&＃xff1b;

job有一下几种状态&＃xff1a;

READY&＃xff0c;需要立即处理的任务&＃xff0c;当延时 (DELAYED) 任务到期后会自动成为当前任务&＃xff1b;
DELAYED&＃xff0c;延迟执行的任务,&＃xff1b;
RESERVED&＃xff0c;已经被消费者获取, 正在执行的任务&＃xff0c;Beanstalkd 负责检查任务是否在 TTR(time-to-run) 内完成&＃xff1b;
BURIED&＃xff0c;保留的任务: 任务不会被执行&＃xff0c;也不会消失&＃xff0c;除非有人将他修改为其他状态&＃xff1b;
DELETED&＃xff0c;消息被彻底删除。Beanstalkd 不再维持这些消息。

状态之间的转移图如下所示&＃xff1a;

思考&＃xff1a;

beanstalkd如何维护job的状态&＃xff1f;tube有3个集合delay、ready和 buried分别存放对应状态的job&＃xff0c;conn的reserved_jobs集合存储状态为reserved的job&＃xff08;消费者获取一个job后&＃xff0c;job的状态才会改变为reserved&＃xff0c;因此这个集合由conn维护&＃xff09;&＃xff1b;
delay状态的job怎么修改为ready&＃xff1f;delay集合是一个按照时间排序的最小堆&＃xff0c;beanstalkd不定时循环从堆根节点获取job&＃xff0c;校验是否需要改变其状态未ready&＃xff1b;
如何实现优先级&＃xff1f;只有ready状态的job才能被消费者获取消费&＃xff0c;ready集合是一个按照优先级排序的最小堆&＃xff0c;根节点始终是优先级最高得job&＃xff1b;
拉模式实现&＃xff1f;消费者使用reserve命令获取job&＃xff0c;beanstalkd检查消费者监听的所有tube&＃xff0c;查找到ready的job即返回&＃xff0c;否则阻塞消费者知道有ready状态的job产生为止&＃xff1b;

2.2beanstalkd命令

beanstalkd支持以下命令&＃xff1a;

3.beanstalkd源码分析

3.1数据结构

3.1.1基础结构体

//堆 struct Heap {int cap; //堆容量int len; //堆元素数目void **data; //元素数组Less less; //元素比较的函数指针Record rec; //函数指针&＃xff0c;将元素插入堆时&＃xff0c;会调用此函数 };//函数指针定义&＃xff1a; typedef int(*Less)(void*, void*); typedef void(*Record)(void*, int);//API&＃xff1a;元素的插入与删除 void * heapremove(Heap *h, int k); int heapinsert(Heap *h, void *x)

//集合 struct ms {size_t used, cap, last; //cap为当前集合容量&＃xff1b;used集合中元素数目&＃xff1b;last上次访问的集合元素的位置void **items; //存储元素的数组ms_event_fn oninsert, onremove; //往集合插入元素&＃xff0c;删除元素时调用的函数 };//函数指针定义如下 typedef void(*ms_event_fn)(ms a, void *item, size_t i);//API void ms_init(ms a, ms_event_fn oninsert, ms_event_fn onremove);//初始化集合 int ms_append(ms a, void *item) //往集合追加元素 int ms_contains(ms a, void *item)//判断集合是否包含元素 void * ms_take(ms a) //获取并删除元素&＃xff08;会从上次访问的位置last开始查找&＃xff09; int ms_remove(ms a, void *item) //删除元素&＃xff0c;从头开始查找 int ms_clear(ms a) //清空集合

3.1.2 管道tube

struct tube {uint refs; //引用计数char name[MAX_TUBE_NAME_LEN]; //名称Heap ready; //存储状态未ready的job&＃xff0c;按照优先级排序Heap delay; //存储状态未delayed的job&＃xff0c;按照到期时间排序struct ms waiting; //等待当前tube有job产生的消费者集合int64 pause; //执行pause命令后&＃xff0c;pause字段记录暂停时间int64 deadline_at; //deadline_at记录暂停到达时间struct job buried; //存储状态为buried的job&＃xff0c;是一个链表 };

创建tube的代码如下&＃xff1a;

tube make_tube(const char *name) {tube t;//底层调用malloc分配空间t &＃61; new(struct tube);if (!t) return NULL;t->name[MAX_TUBE_NAME_LEN - 1] &＃61; &＃39;\0&＃39;;strncpy(t->name, name, MAX_TUBE_NAME_LEN - 1);if (t->name[MAX_TUBE_NAME_LEN - 1] !&＃61; &＃39;\0&＃39;) twarnx("truncating tube name");//设置ready与delay堆的函数指针t->ready.less &＃61; job_pri_less;t->delay.less &＃61; job_delay_less;t->ready.rec &＃61; job_setheappos;t->delay.rec &＃61; job_setheappos;t->buried &＃61; (struct job) { };t->buried.prev &＃61; t->buried.next &＃61; &t->buried;ms_init(&t->waiting, NULL, NULL);return t; }

两个堆指针函数如下&＃xff1a;

//按照优先级比较 int job_pri_less(void *ax, void *bx) {job a &＃61; ax, b &＃61; bx;if (a->r.pri r.pri) return 1;if (a->r.pri > b->r.pri) return 0;return a->r.id r.id; } //按照过期时间比较 int job_delay_less(void *ax, void *bx) {job a &＃61; ax, b &＃61; bx;if (a->r.deadline_at r.deadline_at) return 1;if (a->r.deadline_at > b->r.deadline_at) return 0;return a->r.id r.id; } //设置每个job在堆的index void job_setheappos(void *j, int pos) {((job)j)->heap_index &＃61; pos; }

3.1.3任务job

注&＃xff1a;job创建完成后&＃xff0c;先会保存在全局变量all_jobs的hash表中&＃xff1b;然后才会插入到tube的各job队列中&＃xff1b;

struct job {Jobrec r; // 存储job信息tube tube; //指向其所属tubejob prev, next; //job可以组织为双向链表&＃xff08;buried状态的job就是链表&＃xff09;job ht_next; //所有的job都存储在一个hash表中&＃xff08;拉链法&＃xff09;&＃xff0c;job的id为hash值&＃xff1b;&＃xff08;tube中的job集合存储指针指向各个job&＃xff09;size_t heap_index; /* where is this job in its current heap */…………char body[];//job的数据&＃xff0c;柔性数组 };// job的描述信息 struct Jobrec {uint64 id;uint32 pri;int64 delay;int64 ttr;int32 body_size;int64 created_at; //创建时间int64 deadline_at; //延迟job的过期时间//统计计数uint32 reserve_ct;uint32 timeout_ct;uint32 release_ct;uint32 bury_ct;uint32 kick_ct;byte state;//当前状态 };

3.14套接字socket

struct Socket {int fd;Handle f; //socket发生事件时的处理函数void *x; //服务器监听的socket指向server结构体&＃xff1b;客户端对应的socket指向conn结构体int added; //往epoll注册事件时&＃xff0c;计算操作类型 };

3.15服务器server

struct Server {char *port;char *addr;Socket sock; //监听的socketHeap conns; //存储即将有事件发生的客户端&＃xff1b;按照事件发生的时间排序的最小堆&＃xff1b;//例如&＃xff1a;当客户端获取job后&＃xff0c;若唱过TTR时间没处理完&＃xff0c;job会状态应重置为ready状态&＃xff1b;//当客户端调用reserve获取job但当前tube没有ready状态的job时&＃xff0c;客户端会被阻塞timeout时间&＃xff1b; };

3.1.6客户端链接conn

struct Conn {Server *srv; //执行服务器Socket sock; //客户端socketchar state; //客户端状态&＃xff1a;等待接收命令&＃xff0c;等待接收数据&＃xff0c;等待回复命令&＃xff0c;等待返回job&＃xff0c;关闭&＃xff0c;获取job阻塞中char type; //客户端类型&＃xff1a;生产者&＃xff0c;消费者&＃xff0c;获取job阻塞中Conn *next;tube use; //当前使用的tube&＃xff1b;put命令发布的job会插入到当前tube中int64 tickat; //客户端处理job的TTR到期时间&＃xff1b;或者客户端阻塞的到期时间&＃xff1b;用于在server的conns堆比较int tickpos; // 在srv->conns堆里的位置job soonest_job; //所有reserve任务里到期时间最近的jobint rw; //当前关心的事件: &＃39;r&＃39;, &＃39;w&＃39;, or &＃39;h&＃39;&＃xff08;读、写、关闭连接&＃xff09;int pending_timeout; //客户端获取job而阻塞的到期时间char halfclosed; //表示客户端断开连接char cmd[LINE_BUF_SIZE]; // 输入缓冲区int cmd_len;int cmd_read;char *reply; //输出缓冲区int reply_len;int reply_sent;char reply_buf[LINE_BUF_SIZE];//put命令发布job时&＃xff0c;从客户端读入的jobint in_job_read;job in_job;//待返回给客户端的jobjob out_job;int out_job_sent;//当前客户端监听的所有tube集合struct ms watch;//当前客户端的所有reserved状态的jobstruct job reserved_jobs; };

3.2 服务器启动过程

3.2.1 epoll简介

epoll结构体&＃xff1a;

typedef union epoll_data {void *ptr;int fd;__uint32_t u32;__uint64_t u64; } epoll_data_t;//保存触发事件的某个fd相关的数据struct epoll_event {__uint32_t events; /* epoll event */epoll_data_t data; /* User data variable */ }; //其中events表示感兴趣的事件和被触发的事件&＃xff0c;可能的取值为&＃xff1a; //EPOLLIN&＃xff1a;表示对应的文件描述符可以读&＃xff1b; //EPOLLOUT&＃xff1a;表示对应的文件描述符可以写&＃xff1b; //EPOLLPRI&＃xff1a;表示对应的文件描述符有紧急的数可读&＃xff1b; //EPOLLERR&＃xff1a;表示对应的文件描述符发生错误&＃xff1b; //EPOLLHUP&＃xff1a;表示对应的文件描述符被挂断&＃xff1b;

epoll API定义如下&＃xff1a;

int epoll_create(int size) //生成一个epoll专用的文件描述符&＃xff0c;其中的参数是指定生成描述符的最大范围&＃xff1b;int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event) //用于控制某个文件描述符上的事件&＃xff0c;可以注册事件&＃xff0c;修改事件&＃xff0c;删除事件int epoll_wait(int epfd,struct epoll_event * events,int maxevents,int timeout) //轮询I/O事件的发生&＃xff1b;

3.2.2 beanstalkd使用epoll

//创建epoll&＃xff1a; epfd &＃61; epoll_create(1);//注册事件 int sockwant(Socket *s, int rw) {…………ev.events |&＃61; EPOLLRDHUP | EPOLLPRI;ev.data.ptr &＃61; s; //注意&＃xff1a;传入的是sokcet指针&＃xff1b;&＃xff08;socket的x字段会指向server或者conn结构体&＃xff0c;当socket对应的fd发生事件时&＃xff0c;可以得到server或conn对象&＃xff09;return epoll_ctl(epfd, op, s->fd, &ev); }//等待事件发生 //返回h r w 事件类型 int socknext(Socket **s, int64 timeout) {…………r &＃61; epoll_wait(epfd, &ev, 1, (int)(timeout/1000000));if (r &＃61;&＃61; -1 && errno !&＃61; EINTR) {twarn("epoll_wait");exit(1);}if (r > 0) {*s &＃61; ev.data.ptr; //返回发生事件的socketif (ev.events & (EPOLLHUP|EPOLLRDHUP)) {return &＃39;h&＃39;;} else if (ev.events & EPOLLIN) {return &＃39;r&＃39;;} else if (ev.events & EPOLLOUT) {return &＃39;w&＃39;;}}return 0; }

3.2.3服务器启动

int main(int argc, char **argv) {optparse(&srv, argv&＃43;1);//解析输入参数r &＃61; make_server_socket(srv.addr, srv.port); //创建socketprot_init(); //初始化全局tubes集合&＃xff0c;创建名称为default的默认tubesrvserve(&srv);//启动服务器return 0; }struct ms tubes;//全局变量void prot_init() {//初始化tube集合ms_init(&tubes, NULL, NULL);//创建默认tube&＃xff1b;tube_find_or_make方法会先从tubes集合查找指定名称为tube&＃xff0c;查找到直接返回&＃xff1b;否则创建新的tubeTUBE_ASSIGN(default_tube, tube_find_or_make("default")); }void srvserve(Server *s) {//s->sock为server监听的socket&＃xff1b;设置其处理函数为srvaccept&＃xff1b;s->sock.x &＃61; s;s->sock.f &＃61; (Handle)srvaccept;s->conns.less &＃61; (Less)connless; //设置s->conns堆的函数指针s->conns.rec &＃61; (Record)connrec;r &＃61; listen(s->sock.fd, 1024); //监听r &＃61; sockwant(&s->sock, &＃39;r&＃39;); //注册到epoll//开启循环for (;;) {//服务器有一些事件需要在特定时间执行&＃xff0c;获得最早待执行事件的时间间隔&＃xff0c;作为epoll_wait的等待时间&＃xff1b;后面详细分析函数内部period &＃61; prottick(s);int rw &＃61; socknext(&sock, period); //epoll waitif (rw) {sock->f(sock->x, rw); //调用socket的处理函数}} } //至此&＃xff0c;服务器启动完毕&＃xff0c;等待客户端链接

conns堆分析&＃xff1a;

上面说过&＃xff0c;conns存储即将有事件发生的客户端&＃xff1b;按照事件发生的时间排序的最小堆&＃xff1b;
例如&＃xff1a;当客户端获取job后&＃xff0c;若唱过TTR时间没处理完&＃xff0c;job会状态应重置为ready状态&＃xff1b;
当客户端调用reserve获取job但当前tube没有ready状态的job时&＃xff0c;客户端会被阻塞timeout时间&＃xff1b;

//堆节点比较的函数指针&＃xff1a; int connless(Conn *a, Conn *b) {return a->tickat tickat; }//将客户端对象插入conns堆时&＃xff0c;tickpos记录其插入的index&＃xff08;避免客户端重复插入&＃xff1b;插入之前发现其tickpos>-1则先删除再插入&＃xff09; void connrec(Conn *c, int i) {c->tickpos &＃61; i; }

处理客户端链接请求&＃xff1a;

void srvaccept(Server *s, int ev) {h_accept(s->sock.fd, ev, s); }void h_accept(const int fd, const short which, Server *s) {cfd &＃61; accept(fd, (struct sockaddr *)&addr, &addrlen);flags &＃61; fcntl(cfd, F_GETFL, 0); //获得fd标识r &＃61; fcntl(cfd, F_SETFL, flags | O_NONBLOCK); //设置fd非阻塞&＃xff0c;使用epoll必须设置非阻塞&＃xff0c;负责epoll无法同时监听多个fd//创建conn对象&＃xff1b;默认监听default_tube&＃xff08;c->watch存储所有监听的tube&＃xff09;&＃xff1b;默认使用default_tube&＃xff08;c->use&＃xff09;//注意&＃xff1a;初始化conn对象时&＃xff0c;客户端状态为STATE_WANTCOMMAND&＃xff0c;即等待接收客户端命令&＃xff1b;c &＃61; make_conn(cfd, STATE_WANTCOMMAND, default_tube, default_tube);c->srv &＃61; s;c->sock.x &＃61; c;c->sock.f &＃61; (Handle)prothandle; //设置客户端处理函数c->sock.fd &＃61; cfd;r &＃61; sockwant(&c->sock, &＃39;r&＃39;); //epoll注册&＃xff0c;监听可读事件 }

当客户端socket可读或可写时&＃xff0c;会执行prothandle函数&＃xff1a;

static void prothandle(Conn *c, int ev) {h_conn(c->sock.fd, ev, c); }static void h_conn(const int fd, const short which, Conn *c) {//客户端断开链接&＃xff0c;标记if (which &＃61;&＃61; &＃39;h&＃39;) {c->halfclosed &＃61; 1;}//客户端数据交互&＃xff08;根据客户端状态不同执行不同的读写操作&＃xff09;conn_data(c);//解析完命令时&＃xff0c;执行命令while (cmd_data_ready(c) && (c->cmd_len &＃61; cmd_len(c))) do_cmd(c);}

3.3 服务器与客户端的数据交互

beanstalkd将客户端conn分为以下几种状态&＃xff1a;等待接受命令&＃xff0c;等待接收数据&＃xff0c;等待回复数据&＃xff0c;等待返回job等&＃xff1b;

#define STATE_WANTCOMMAND 0 #define STATE_WANTDATA 1 #define STATE_SENDJOB 2 #define STATE_SENDWORD 3 #define STATE_WAIT 4 #define STATE_BITBUCKET 5 #define STATE_CLOSE 6

当客户端fd可读或者可写时&＃xff0c;服务器根据当前客户端的状态执行不同的操作&＃xff1a;

注意&＃xff1a;TCP是基于流的&＃xff0c;因此存在半包、粘包问题&＃xff1b;即&＃xff0c;服务器一次read的命令请求数据可能不完整&＃xff0c;或者一次read多个命令请求的数据&＃xff1b;

//有些状态操作已省略 static void conn_data(Conn *c) {switch (c->state) {case STATE_WANTCOMMAND:r &＃61; read(c->sock.fd, c->cmd &＃43; c->cmd_read, LINE_BUF_SIZE - c->cmd_read); //读取命令到输入缓冲区cmdc->cmd_read &＃43;&＃61; r;c->cmd_len &＃61; cmd_len(c); //定位\r\n&＃xff0c;并返回命令请求开始位置到\r\n长度&＃xff1b;如果没有\r\b说明命令请求不完全&＃xff0c;返回0if (c->cmd_len) return do_cmd(c); //如果读取完整的命令&＃xff0c;则处理&＃xff1b;否则意味着命令不完全&＃xff0c;需要下次继续接收break;case STATE_WANTDATA: //只有当使用put命令发布任务时&＃xff0c;才会携带数据&＃xff1b;客户端状态才会成为STATE_WANTDATA&＃xff1b;//而读取命令行时&＃xff0c;已经携带了任务的必要参数&＃xff0c;那时已经创建了任务&＃xff0c;并存储在c->in_job字段j &＃61; c->in_job;r &＃61; read(c->sock.fd, j->body &＃43; c->in_job_read, j->r.body_size -c->in_job_read); //读取任务数据c->in_job_read &＃43;&＃61; r; //记录任务读取了多少数据maybe_enqueue_incoming_job(c); //函数会判断任务数据是否已经读取完全&＃xff0c;完全则将任务写入tube的ready或delay队列&＃xff1b;后面会将break;case STATE_SENDWORD: //回复客户端命令请求r&＃61; write(c->sock.fd, c->reply &＃43; c->reply_sent, c->reply_len - c->reply_sent);c->reply_sent &＃43;&＃61; r; //已经发送的字节数if (c->reply_sent &＃61;&＃61; c->reply_len) return reset_conn(c); //如果返回数据已经发完&＃xff0c;则重置客户端rw&＃xff0c;关心可读事件&＃xff1b;否则继续待发送数据状态break;case STATE_SENDJOB: //待发送jobj &＃61; c->out_job;//返回数据与jobiov[0].iov_base &＃61; (void *)(c->reply &＃43; c->reply_sent);iov[0].iov_len &＃61; c->reply_len - c->reply_sent; /* maybe 0 */iov[1].iov_base &＃61; j->body &＃43; c->out_job_sent;iov[1].iov_len &＃61; j->r.body_size - c->out_job_sent;r &＃61; writev(c->sock.fd, iov, 2);c->reply_sent &＃43;&＃61; r;if (c->reply_sent >&＃61; c->reply_len) {c->out_job_sent &＃43;&＃61; c->reply_sent - c->reply_len;c->reply_sent &＃61; c->reply_len;}if (c->out_job_sent &＃61;&＃61; j->r.body_size) { //如果job的数据已经发完&＃xff0c;则重置客户端rw&＃xff0c;关心可读事件&＃xff1b;否则继续待发送jobreturn reset_conn(c);}break;} }

3.4 命令的处理过程

3.4.1查找命令

//命令执行的入口函数 static void do_cmd(Conn *c) {dispatch_cmd(c); //分发并执行命令fill_extra_data(c); //put命令时&＃xff0c;不仅需要执行命令&＃xff0c;还需要接续job数据 }static void dispatch_cmd(Conn *c) {//查找命令类型type &＃61; which_cmd(c);//switch处理各个命令switch (type) {…………} }

beanstalkd有以下命令定义&＃xff1a;

//命令字符串 #define CMD_PUT "put " #define CMD_PEEK_READY "peek-ready" #define CMD_RESERVE "reserve" #define CMD_RELEASE "release " …………//命令编码类型&＃xff1a; #define OP_UNKNOWN 0 #define OP_PUT 1 #define OP_PEEKJOB 2 #define OP_RESERVE 3 #define OP_DELETE 4 #define OP_RELEASE 5 …………

查找命令其实就是字符串比较&＃xff1a;

static int which_cmd(Conn *c) {//宏定义&＃xff1b;比较输入缓冲区命令字符串与命令表中字符串比较&＃xff0c;返回命令类型#define TEST_CMD(s,c,o) if (strncmp((s), (c), CONSTSTRLEN(c)) &＃61;&＃61; 0) return (o);//宏替换后就是一系列if语句TEST_CMD(c->cmd, CMD_PUT, OP_PUT);TEST_CMD(c->cmd, CMD_PEEKJOB, OP_PEEKJOB);………… }

3.4.2命令1——发布任务

case OP_PUT:r &＃61; read_pri(&pri, c->cmd &＃43; 4, &delay_buf); //解析优先级prir &＃61; read_delay(&delay, delay_buf, &ttr_buf); //解析delayr &＃61; read_ttr(&ttr, ttr_buf, &size_buf); //解析ttrbody_size &＃61; strtoul(size_buf, &end_buf, 10); //解析job字节数op_ct[type]&＃43;&＃43;; //统计if (body_size > job_data_size_limit) { //job长度超过限制&＃xff1b;返回return skip(c, body_size &＃43; 2, MSG_JOB_TOO_BIG);}//put&＃xff0c;说明是生产者&＃xff0c;设置conn类型为生产者connsetproducer(c);//初始化job结构体&＃xff0c;存储在hash表all_jobs中c->in_job &＃61; make_job(pri, delay, ttr, body_size &＃43; 2, c->use);//解析客户端发来的任务数据&＃xff0c;存储在c->in_job的body数据字段fill_extra_data(c);//校验job数据是否读取完毕&＃xff0c;完了则入tube的队列maybe_enqueue_incoming_job(c);

任务入队列&＃xff1a;

static void maybe_enqueue_incoming_job(Conn *c) {job j &＃61; c->in_job;//任务数据已经读取完毕&＃xff0c;入队列&＃xff08;ready或者delay队列&＃xff09;if (c->in_job_read &＃61;&＃61; j->r.body_size) return enqueue_incoming_job(c);//任务数据没有读取完毕&＃xff0c;则设置客户端conn状态未等待接收数据STATE_WANTDATAc->state &＃61; STATE_WANTDATA; }static void enqueue_incoming_job(Conn *c) {int r;job j &＃61; c->in_job;c->in_job &＃61; NULL; /* the connection no longer owns this job */c->in_job_read &＃61; 0;//入队列r &＃61; enqueue_job(c->srv, j, j->r.delay, 1);//返回数据&＃xff1b;并设置conn状态为STATE_SENDWORDreply_line(c, STATE_SENDWORD, MSG_BURIED_FMT, j->r.id); }static int enqueue_job(Server *s, job j, int64 delay, char update_store) {int r;j->reserver &＃61; NULL;if (delay) { //入delay队列&＃xff0c;设置任务的deadline_atj->r.deadline_at &＃61; nanoseconds() &＃43; delay;r &＃61; heapinsert(&j->tube->delay, j);j->r.state &＃61; Delayed;} else { //入ready队列r &＃61; heapinsert(&j->tube->ready, j);if (!r) return 0;j->r.state &＃61; Ready; }//检查有没有消费者正在阻塞等待此tube产生job&＃xff0c;若有需要返回job&＃xff1b;process_queue();return 1; }

返回命令回复给客户端&＃xff1a;

//reply_line函数组装命令回复数据&＃xff0c;调用reply函数&＃xff1b;只是将数据写入到输出缓冲区&＃xff0c;并修改了客户端状态为STATE_SENDWORD&＃xff0c;实际发送数据在3.3节已经说过&＃xff1b; static void reply(Conn *c, char *line, int len, int state) {if (!c) return;connwant(c, &＃39;w&＃39;);//修改关心的事件为可写事件c->next &＃61; dirty; //放入dirty链表dirty &＃61; c;c->reply &＃61; line; //输出数据缓冲区c->reply_len &＃61; len;c->reply_sent &＃61; 0;c->state &＃61; state; //设置conn状态 }

connwant函数实现如下&＃xff1a;

void connwant(Conn *c, int rw) {c->rw &＃61; rw; //c->rw记录当前客户端关心的socket事件connsched(c); }void connsched(Conn *c) {if (c->tickpos > -1) { //c->tickpos记录当前客户端在srv->conns堆的索引&＃xff1b;&＃xff08;思考&＃xff1a;tickpos在什么时候赋值的&＃xff1f;heap的函数指针rec&＃xff09;heapremove(&c->srv->conns, c->tickpos);}c->tickat &＃61; conntickat(c); //计算当前客户端待发生的某个事件的时间if (c->tickat) {heapinsert(&c->srv->conns, c); //插入srv->conns堆} }

问题1&＃xff1a;connwant只是修改了conn的rw字段为‘w’&＃xff0c;表示关心客户端的读时间&＃xff0c;什么时候调用epoll注册呢&＃xff1f;dirty链表又是做什么的呢&＃xff1f;

beanstalkd有个函数update_conns负责更新客户端socket的事件到epoll&＃xff1b;其在每次循环开始&＃xff0c;执行epoll_wait之前都会执行&＃xff1b;

static void update_conns() {int r;Conn *c;while (dirty) { //遍历dirty链表&＃xff0c;更新每一个conn关心的socket事件c &＃61; dirty;dirty &＃61; dirty->next;c->next &＃61; NULL;r &＃61; sockwant(&c->sock, c->rw);if (r &＃61;&＃61; -1) {twarn("sockwant");connclose(c);}} }

问题2&＃xff1a;srv->conns存储的客户端都是在某个时间点有事件待处理的&＃xff0c;客户端都有哪些事件需要处理呢&＃xff1f;

1&＃xff09;消费者获取job后&＃xff0c;job的状态改为reserved&＃xff0c;当TTR时间过后&＃xff0c;如果客户端还没有处理完这个job&＃xff0c;服务器会将这个job的状态重置为ready&＃xff0c;以便让其他消费者可以消费&＃xff1b;
2&＃xff09;消费者调用reserve获取job时&＃xff0c;假如其监听的tube没有ready状态的job&＃xff0c;那么客户端将会被阻塞&＃xff0c;直到有job产生&＃xff0c;或者阻塞超时&＃xff1b;

//计算当前客户端待处理事件的deadline static int64 conntickat(Conn *c) {//客户端正在阻塞if (conn_waiting(c)) {margin &＃61; SAFETY_MARGIN;}//如果客户端有reserved状态的任务&＃xff0c;则获取到期时间最近的&＃xff1b;&＃xff08;当客户端处于阻塞状态时&＃xff0c;应该提前SAFETY_MARGIN时间处理此事件&＃xff09;//connsoonestjob&＃xff1a;获取到期时间最近的reserved jobif (has_reserved_job(c)) {t &＃61; connsoonestjob(c)->r.deadline_at - nanoseconds() - margin;should_timeout &＃61; 1;}//客户端阻塞超时时间if (c->pending_timeout >&＃61; 0) {t &＃61; min(t, ((int64)c->pending_timeout) * 1000000000);should_timeout &＃61; 1;}//返回时间发生的时间&＃xff1b;后续会将此客户端插入srv->conns堆&＃xff0c;且是按照此时间排序的&＃xff1b;if (should_timeout) {return nanoseconds() &＃43; t;}return 0; }

问题3&＃xff1a;当生产者新发布一个job到某个tube时&＃xff0c;此时可能有其他消费者监听此tube&＃xff0c;且阻塞等待job的产生&＃xff0c;此时就需要将此job返回给消费者&＃xff1b;处理函数为process_queue

static void process_queue() {int64 now &＃61; nanoseconds();while ((j &＃61; next_eligible_job(now))) { //遍历所有tube&＃xff0c;当tube有客户端等待&＃xff0c;且有ready状态的job时&＃xff0c;返回jobheapremove(&j->tube->ready, j->heap_index);//ms_take&＃xff1a;将客户端从此job所属tube的waiting集合中删除&＃xff1b;并返回客户端conn//remove_waiting_conn&＃xff1a;从当前客户端conn监听的所有tube的waiting队列中移除自己//reserve_job&＃xff1a;返回此job给客户端reserve_job(remove_waiting_conn(ms_take(&j->tube->waiting)), j);} }static job next_eligible_job(int64 now) {tube t;size_t i;job j &＃61; NULL, candidate;//循环所有tubefor (i &＃61; 0; i pause) { //假如tube正在暂停&＃xff0c;且超时时间未到&＃xff0c;则跳过if (t->deadline_at > now) continue;t->pause &＃61; 0;}if (t->waiting.used && t->ready.len) { //tube的waiting集合有元素说明有客户端正在阻塞等待此tube产生任务&＃xff1b;有ready状态的任务candidate &＃61; t->ready.data[0]; //从tubes里获取满足条件的优先级最高得job返回if (!j || job_pri_less(candidate, j)) {j &＃61; candidate;}}}return j; }Conn * remove_waiting_conn(Conn *c) {tube t;size_t i;if (!conn_waiting(c)) return NULL;c->type &&＃61; ~CONN_TYPE_WAITING; //去除CONN_TYPE_WAITING标志global_stat.waiting_ct--;for (i &＃61; 0; i watch.used; i&＃43;&＃43;) { //遍历客户端监听的所有tube&＃xff0c;挨个从tube的waiting队列中删除自己t &＃61; c->watch.items[i];t->stat.waiting_ct--;ms_remove(&t->waiting, c);}return c; }static void reserve_job(Conn *c, job j) {j->r.deadline_at &＃61; nanoseconds() &＃43; j->r.ttr; //job的实效时间j->r.state &＃61; Reserved; //状态改为Reservedjob_insert(&c->reserved_jobs, j); //插入客户端的reserved_jobs链表j->reserver &＃61; c; //记录job当前消费者if (c->soonest_job && j->r.deadline_at soonest_job->r.deadline_at) { //soonest_job记录最近要到期的Reserved状态的job&＃xff0c;更新&＃xff1b;c->soonest_job &＃61; j;}return reply_job(c, j, MSG_RESERVED); //返回job }

3.4.3 命令2——获取任务reserve

case OP_RESERVE_TIMEOUT:timeout &＃61; strtol(c->cmd &＃43; CMD_RESERVE_TIMEOUT_LEN, &end_buf, 10); //reserve可以设置阻塞超时时间&＃xff0c;解析case OP_RESERVE:op_ct[type]&＃43;&＃43;;connsetworker(c); //设置客户端类型为消费者CONN_TYPE_WORKER//当客户端有多个任务正在处理&＃xff0c;处于reserved状态&＃xff0c;且超时时间即将到达时&＃xff1b;如果此时客户端监听的所有tube都没有ready状态的任务&＃xff0c;则直接返回MSG_DEADLINE_SOON给客户端if (conndeadlinesoon(c) && !conn_ready(c)) {return reply_msg(c, MSG_DEADLINE_SOON);}//设置当前客户端正在等待jobwait_for_job(c, timeout);//同3.4.2节process_queue();

上面说过&＃xff0c;当客户端有多个任务正在处理&＃xff0c;处于reserved状态&＃xff0c;且超时时间即将到达时&＃xff1b;

如果此时客户端监听的所有tube都没有ready状态的任务&＃xff0c;则直接返回MSG_DEADLINE_SOON给客户端&＃xff1b;

否则会导致客户端的阻塞&＃xff0c;导致这些reserved的任务超时&＃xff1b;

static void wait_for_job(Conn *c, int timeout) {c->state &＃61; STATE_WAIT; //设置客户端状态为STATE_WAITenqueue_waiting_conn(c); //将客户端添加到其监听的所有tube的waiting队列中//设置客户端的超时时间c->pending_timeout &＃61; timeout;//修改关心的事件为可读事件connwant(c, &＃39;h&＃39;);c->next &＃61; dirty; //将当前客户端添加到dirty链表中dirty &＃61; c; }static void enqueue_waiting_conn(Conn *c) {tube t;size_t i;global_stat.waiting_ct&＃43;&＃43;;c->type |&＃61; CONN_TYPE_WAITING;for (i &＃61; 0; i watch.used; i&＃43;&＃43;) { //c->watch为客户端监听的所有tubet &＃61; c->watch.items[i];t->stat.waiting_ct&＃43;&＃43;;ms_append(&t->waiting, c); //t->waiting为等待当前tube有任务产生的所有客户端} }

3.4.4 循环之始epoll_wait之前

在执行epoll_wait之前&＃xff0c;需要计算超时时间&＃xff1b;不能被epoll_wait一直阻塞&＃xff1b;服务器还有很多事情待处理&＃xff1b;

1&＃xff09;将状态未delay的且已经到期的job移到ready队列&＃xff1b;
2&＃xff09;tube暂停时间到达&＃xff0c;如果tube存在消费者阻塞等待获取job&＃xff0c;需要返回job给客户端&＃xff1b;
3&＃xff09;消费者消费的状态为reserved的job可能即将超时到期&＃xff1b;
4&＃xff09;客户端阻塞等待job的超时时间可能即将达到&＃xff1b;

服务器需要及时处理这些所有事情&＃xff0c;因此epoll_wait等待时间不能过长&＃xff1b;

int64 prottick(Server *s) {int64 period &＃61; 0x34630B8A000LL; //默认epoll_wait等待时间now &＃61; nanoseconds();while ((j &＃61; delay_q_peek())) { //遍历所有tube的delay队列中过期时间已经到达或者即将的job&＃xff08;即将到达时间最小&＃xff09;d &＃61; j->r.deadline_at - now;if (d > 0) {period &＃61; min(period, d); //即将到达&＃xff0c;更新periodbreak;}j &＃61; delay_q_take();r &＃61; enqueue_job(s, j, 0, 0); //job入队到ready队列if (r <1) bury_job(s, j, 0); /* out of memory, so bury it */}for (i &＃61; 0; i deadline_at - now;if (t->pause && d <&＃61; 0) { //tube暂停期限达到&＃xff0c;process_queue同3.4.2节t->pause &＃61; 0;process_queue();}else if (d > 0) {period &＃61; min(period, d); //tube暂停即将到期&＃xff0c;更新period}}while (s->conns.len) {Conn *c &＃61; s->conns.data[0]; //循环获取conn待执行事件发生时间最早的d &＃61; c->tickat - now;if (d > 0) { //发生事件未到&＃xff0c;更新period&＃xff0c;结束循环period &＃61; min(period, d);break;}heapremove(&s->conns, 0); //否则&＃xff0c;移除conn&＃xff0c;处理客户端事件conn_timeout(c);}update_conns(); //更新客户端关心的socke事件&＃xff0c;其实就是遍历dirty链表return period; }static job delay_q_peek() {int i;tube t;job j &＃61; NULL, nj;for (i &＃61; 0; i delay.len &＃61;&＃61; 0) {continue;}nj &＃61; t->delay.data[0];if (!j || nj->r.deadline_at r.deadline_at) j &＃61; nj;}return j; }static void conn_timeout(Conn *c) {int r, should_timeout &＃61; 0;job j;//客户端正在被阻塞时&＃xff0c;如果有reserved状态的job即将到期&＃xff0c;则需要解除客户端阻塞//conndeadlinesoon&＃xff1a;查询到期时间最小的reserved job&＃xff0c;校验其是否即将到期&＃xff08;1秒内到期&＃xff09;if (conn_waiting(c) && conndeadlinesoon(c)) should_timeout &＃61; 1;//connsoonestjob获取到期时间最近的reserved jobwhile ((j &＃61; connsoonestjob(c))) {if (j->r.deadline_at >&＃61; nanoseconds()) break;timeout_ct&＃43;&＃43;; //已经超时j->r.timeout_ct&＃43;&＃43;;r &＃61; enqueue_job(c->srv, remove_this_reserved_job(c, j), 0, 0); //从客户端的reserved_jobs链表移除job&＃xff0c;重新入到tube的相应job队列if (r <1) bury_job(c->srv, j, 0); /* out of memory, so bury it */connsched(c); //重新计算conn待处理事件的时间&＃xff0c;入srv->conns堆}if (should_timeout) {return reply_msg(remove_waiting_conn(c), MSG_DEADLINE_SOON); //reserved即将到期&＃xff0c;解除阻塞&＃xff0c;返回MSG_DEADLINE_SOON消息} else if (conn_waiting(c) && c->pending_timeout >&＃61; 0) { //客户端阻塞超时&＃xff0c;解除阻塞c->pending_timeout &＃61; -1;return reply_msg(remove_waiting_conn(c), MSG_TIMED_OUT);} }