Beanstalkd 源码初探

引言

Beanstalkd 是一个比较轻量级的消息队列服务，对于性能和稳定性要求不是特别高（相对于 RabbitMQ, Redis, Kafka 等），并且需要延迟执行任务的场景非常合适；此外，它也支持给任务设置不同的优先级、执行超时时间等。

在我们的业务中，经常会借助 Beanstalkd 执行队列任务，常见的用例如下：

1. 用户完成会员购买并激活后，发送私信通知、重置账号重命名状态等；
2. 用户完成评论后，异步更新评论计数；
3. 用户对私家课收听记录上报后，异步更新最近收听的小节、累积收听时长、同步到其它系统等；
4. 用户记录添加后，会同步至 Redis，为保证数据库和 Redis 的数据最终一致性，会提前启动一个延迟校验的任务（如 5s 后），检查 Redis 中与数据库记录是否一致。

Beanstalkd 初识

特点

1. 基于 TCP 并采用 ASCII 编码的文本协议。详细定义参见：protocol.txt：
   1. 客户端负责与服务端的交互：连接、发送命令和数据、等待响应、关闭连接
   2. 服务端串行处理每个客户端连接
   3. 协议由两个部分组成：文本行（用于客户端命令和服务端响应）和非结构化的数据块（用于传送任务 body 和 stats 信息）

1. 队列消息是存储在内存中的，但用户可以选择开启 WAL 机制（binlog），这样重启后可以回放任务，提高了可用性
2. 采用类似 Redis 的单线程模型（IO 多路复用机制），因此不必考虑多线程环境下线程同步、加锁等，简化实现

关键词

1. Tube：类似 Kafka 中的 Topic，或者其它队列系统中的 Channel
2. Job：客户端生产和消费的基本单元。每个任务都有特定的 id，可设定优先级，超时时间，延迟执行时间等
3. WAL (Write Ahead Log)：负责 binlog 管理（写入、压缩、日志文件清理、任务恢复等）
4. Server：Beanstalkd 服务端
5. Conn：Beanstalkd 客户端连接处理

任务状态流转

任务典型生命周期

工作方式描述

1. 服务端会有一到多个 tubes（在数组中维护）。每个 tube 都会包含一个就绪队列（在最小堆维护）以及一个延迟队列（也在最小堆维护）。每个任务都会在一个特定的 tube 中度过全部的生命周期
2. 客户端可以使用 watch 指令订阅某个 tube，也可以使用 ignore 取消订阅，消费者可以同时订阅多个 tube，当消费者 reserve 任务时，该任务可能来自其 watch list 中的任意一个 tube
3. 当客户端连接时，默认会使用 default tube，可以使用 use 切换 tube
4. tube 是会根据需要随时创建的，当没有客户端引用时，就会被删除

安装

借助 Docker 启动一个 Beanstalkd 服务非常轻松，请运行下面的命令行即可：

docker run -d -p 11300:11300 schickling/beanstalkd

如果上述命令行执行正常，则 Beanstalkd 服务应该启动了，其默认监听的端口号为 11300，运行 docker ps 可以查看服务是否正常启动并运行：

编译 & 运行 & 调试

首先，需要前往 beanstalkd 仓库克隆 Master 分支源码至本地。

为了方便管理 C 项目，这里使用了 JET BRAINS 家族的 Clion。当然，你也可以使用自己喜欢的工具打开。

由于 Clion 使用了 CMake 管理 C&C++ 项目，所以打开项目时需要在其根目录下创建一个 CMakeLists.txt 文件，并填写如下内容：

cmake_minimum_required(VERSION 3.13)
project(beanstalkd C)
set(BUILD_DIR .)
add_custom_target(beanstalkd ALL COMMAND make WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
add_custom_command(TARGET beanstalkd POST_BUILD
        COMMAND echo copy ${PROJECT_NAME} to ${CMAKE_CURRENT_BINARY_DIR}
        COMMAND cp ${CMAKE_CURRENT_SOURCE_DIR}/beanstalkd ${CMAKE_CURRENT_BINARY_DIR}
)

🎉 至此，准备工作已经做完啦。接下来，可以尝试点击「构建」按钮，进行编译。编译结束后，就可以点击运行启动 Beanstalkd 服务啦。哦，对了，如果需要调试支持的话，直接在需要的地方打上断点，并点击「调试」按钮即可开始。

关于 Makefile

查看 Makefile 文件，可以看到有如下几个命令可以执行：

1. make all: 编译、链接并生成可执行的二进制文件 beanstalkd。由于我们已经将该命令放到 CMakeLists.txt 文件中，在使用 Clion 构建时可自动触发
2. make install: 将生成的可执行文件 beanstalkd 安装到 BINDIR=$(DESTDIR)/usr/local/bin 目录下
3. make clean: 清理生成的 *.o 文件
4. make bench: 跑 Benchmark 用

客户端使用示例

下面看一个简单的例子。生产者负责将一组待抓取的 URLs 放到队列中，再由一组消费者并发访问队列中的 URLs。主流程的示例代码如下：

// 会首先启动 NUM_WORKERS 个消费者在不同的线程中监听
// 然后让生产者向队列中填充 URLs，供消费者使用
fn main() {
    const NUM_WORKERS: isize = 5;

    let mut handles = vec![];
    for i in 0..NUM_WORKERS {
        // 启动 NUM_WORKERS 个消费者
        let hd = thread::spawn(move || consume_urls(i));
        handles.push(hd);
    }

    produce_urls();

    // 等待消费者结束
    for hd in handles {
        hd.join().unwrap();
    }
}

生产者

fn produce_urls() {
    let mut client = Beanstalkd::localhost().unwrap();
    client.tube("urls").unwrap();

    let urls = vec![
        "https://github.com/iFaceless/ifaceless.github.io",
        "https://github.com/iFaceless/gic",
        "https://github.com/iFaceless/rust-exercises",
        "https://github.com/iFaceless/learning-rust",
        "https://github.com/iFaceless/fixture",
        "https://github.com/iFaceless/rest",
        "https://github.com/iFaceless/bigcache",
        "https://github.com/iFaceless/leetgogo",
        "https://github.com/iFaceless/freecache",
    ];

    for url in urls {
        client.put(url, 0, 0, 1000).unwrap();
    }
}

消费者

fn consume_urls(id: isize) {
    println!("[Consumer {}] started...", id);
    let mut client = Beanstalkd::localhost().unwrap();
    client.watch("urls").unwrap();

    loop {
        let (job_id, url) = match client.reserve() {
            Ok(job) => job,
            Err(e) => {
                println!("[Consumer {}] error happens: {}", id, e);
                break;
            }
        };

        println!("[Consumer {}] got job <{}>: {}", id, job_id, url);
        client.delete(job_id).unwrap();
    }
}

源码探索

模块分类图

为了方便阅读源码，粗略地根据自己的理解给各个文件做了简单的分类：

模块 UML 图

虽说 Beanstalkd 的源码是使用 C 编写的，但是其中的设计思想依然可以从面向对象的角度来解释。比如模块化设计、接口设计、多态等。根据自己的理解，对其中的一些核心模块做了梳理，并绘制了一个简单的 UML 图来加深理解：

基本数据结构

最小堆

二叉堆（Heap） 是一种很常见的数据结构，本质上是一棵完全二叉树。其分为最大堆（也叫大根堆） 和 最小堆（也叫小根堆）：

1. 最大堆：根结点的键值是所有堆结点键值中最大者的堆
2. 最小堆：根结点的键值是所有堆结点键值中最小者的堆

在 beanstalkd/heap.c 是对最小堆的实现。那么，beanstalkd 中哪些地方用到了最小堆呢？

1. Tube 中的延迟任务队列（最先到期的任务会在堆顶，这样可以在 O(1) 时间复杂度获取到）
2. Tube 中的就绪任务队列（基于优先级排列，优先级最高的任务会在堆顶）
3. Server 中的客户端连接队列（基于 tickat 时间排列）

接下来，我们看看最小堆的实现：

int
heapinsert(Heap *h, void *x)
{
    int k;

    // 扩容策略：2 倍长度
    if (h->len == h->cap) {
        void **ndata;
        int ncap = (h->len+1) * 2; /* allocate twice what we need */

        ndata = malloc(sizeof(void*) * ncap);
        if (!ndata) {
            return 0;
        }

        memcpy(ndata, h->data, sizeof(void*)*h->len);
        free(h->data);
        // 指向新的位置
        h->data = ndata;
        // 更新容量
        h->cap = ncap;
    }

    k = h->len;
    h->len++;
    set(h, k, x);
    siftdown(h, k);
    return 1;
}

void *
heapremove(Heap *h, int k)
{
    void *x;

    if (k >= h->len) {
        return 0;
    }

    x = h->data[k];
    h->len--;
    // 用原来的数组最后一位覆盖被删除的位置
    set(h, k, h->data[h->len]);
    siftdown(h, k);
    siftup(h, k);
    h->rec(x, -1);
    return x;
}

static void
set(Heap *h, int k, void *x)
{
    h->data[k] = x;
    // 这里会去调用相应的回调函数
    h->rec(x, k);
}

// 判断位置 a 指向的对象是否小于 b 指向的对象
static int
less(Heap *h, int a, int b)
{
    // h->less 是一个判断大小的回调
    // 其实要在面向对象的语言中，完全可以自定一个实现了比较大小的接口
    // 比如在 Rust 中，可以使用 `PartialEq` 限定...
    return h->less(h->data[a], h->data[b]);
}

变长数组

在 C 语言标准库中是没有可变长度的数组实现的，所以在 beanstalkd/ms.c 实现了一种类似的数据结构，它具有如下特点：

1. ms 结构体维护一个可动态扩容的数组（**items）
2. 扩容策略很粗暴，直接扩充为原来容量的两倍
3. 插入的平均时间复杂度为 O(1)
4. 删除的平均时间复杂度为 O(1)
5. 由于删除时，会将尾部 item 替换掉被删除的 item，所以不能依赖数组中的元素顺序（顺序不保证和添加时一致）
6. 删除 item 后，其实数组占用的内存空间还在（并没有动态缩容的策略）

那具体在哪些地方用到了 ms 这种数据结构呢？梳理后，主要发现以下几处：

1. 全局的 Tube 列表
2. 客户端连接的 Conn 中维护的 watch list
3. 与 Tube 关联的等待连接（conns）列表

下面看看其具体的实现：

// 初始化数组，并注册插入和移除的回调函数
void
ms_init(ms a, ms_event_fn oninsert, ms_event_fn onremove)
{
    a->used = a->cap = a->last = 0;
    a->items = NULL;
    a->oninsert = oninsert;
    a->onremove = onremove;
}

// 控制数组增长
static void
grow(ms a)
{
    void **nitems;
    // 倍速增长：1, 2, 4, ...
    size_t ncap = (a->cap << 1) ? : 1;

    nitems = malloc(ncap * sizeof(void *));
    if (!nitems) return;

    // 旧的数据拷贝到新开辟的空间
    memcpy(nitems, a->items, a->used * sizeof(void *));
    // 释放旧的内存空间
    free(a->items);
    // 指向新的位置
    a->items = nitems;
    // 更新数组容量
    a->cap = ncap;
}

// 在数组尾部插入新的 item
// O(1)
int
ms_append(ms a, void *item)
{
    // 按需扩展容量
    if (a->used >= a->cap) grow(a);
    // 扩容失败，就返回，也就是不能再新增 item 了
    if (a->used >= a->cap) return 0;

    a->items[a->used++] = item;
    // 如果有回调，则触发回调函数
    if (a->oninsert) a->oninsert(a, item, a->used - 1);
    return 1;
}

// 删除指定位置的 item
// O(1)
static int
ms_delete(ms a, size_t i)
{
    void *item;

    if (i >= a->used) return 0;
    item = a->items[i];
    // 相当于把尾部 item 写到被删除的位置，并「缩容」
    a->items[i] = a->items[--a->used];

    /* it has already been removed now */
    if (a->onremove) a->onremove(a, item, i);
    return 1;
}

字典

在 beanstalkd/job.c 中，为了方便基于 job_id 快速定位到具体的任务，作者实现了一个字典数据结构。这里是和 job 耦合在一起实现的，根据对源码的分析，可以得出该字典数据结构的特点如下：

1. 采用基于 job_id 哈希取模的方式计算 slot_id
2. 使用链地址法解决哈希冲突
3. 根据负载因子自动进行 rehash（进行扩容或缩容），扩容或者缩容的系数根据 beanstalkd/primes.c 设置
4. rehash 过程并没有采用类似 Redis 中渐进式 rehash 机制，而是阻塞式完成整个哈希表的 rehash 后才可以进行后续操作

存放 job 及 rehash 的详细源码分析如下：

// 存放一个 job
static void
store_job(job j)
{
    int index = 0;

    index = _get_job_hash_index(j->r.id);

    j->ht_next = all_jobs[index]; // 如果存在冲突，就采用链地址法
    all_jobs[index] = j;
    all_jobs_used++;

    /* accept a load factor of 4 */
    // 负载因子设置为 4，超过阈值时就会进行 rehash
    // 看起这里是阻塞的方式来进行 rehash 了，如果 hash 表太大，会被阻塞
    // 并没有使用类似 redis 那样渐进式 rehash 思路
    if (all_jobs_used > (all_jobs_cap << 2)) rehash(1);
}

// 支持扩容和缩容
static void
rehash(int is_upscaling)
{
    job *old = all_jobs;
    // 记录下旧的 hash 表容量，元素个数
    size_t old_cap = all_jobs_cap, old_used = all_jobs_used, i;
    int old_prime = cur_prime;
    int d = is_upscaling ? 1 : -1;

    if (cur_prime + d >= NUM_PRIMES) return;
    if (cur_prime + d < 0) return;
    if (is_upscaling && hash_table_was_oom) return;

    cur_prime += d;

    all_jobs_cap = primes[cur_prime];
    all_jobs = calloc(all_jobs_cap, sizeof(job));
    if (!all_jobs) { // 针对扩容失败的处理，恢复原来的不变，但是标记 OOM
        twarnx("Failed to allocate %zu new hash buckets", all_jobs_cap);
        hash_table_was_oom = 1;
        cur_prime = old_prime;
        all_jobs = old;
        all_jobs_cap = old_cap;
        all_jobs_used = old_used;
        return;
    }
    // 重置 hash 表状态
    all_jobs_used = 0;
    hash_table_was_oom = 0;

    // 其实就是把 Hash 表上所有的 jobs 全部映射到新的空间
    for (i = 0; i < old_cap; i++) {
        while (old[i]) {
            job j = old[i];
            old[i] = j->ht_next;
            j->ht_next = NULL;
            store_job(j);
        }
    }

    // 然后把原来的内存空间释放掉
    if (old != all_jobs_init) {
        free(old);
    }
}

链表

链表这种数据结构在 Beanstalkd 实现中用得比较频繁，比如

1. beanstalkd/walg.c 中使用了单向链表的串联了一些列的日志文件（参见三个游标指针：head, cur, tail）
2. beanstalkd/conn.c 使用双向链表连接了一些列被 reserve 的任务

在 beastalkd/job.c，可以看到任务双向链表实现：

int
job_list_any_p(job head)
{
    return head->next != head || head->prev != head;
}

job
job_remove(job j)
{
    if (!j) return NULL;
    if (!job_list_any_p(j)) return NULL; /* not in a doubly-linked list */

    j->next->prev = j->prev;
    j->prev->next = j->next;

    j->prev = j->next = j;

    return j;
}

void
job_insert(job head, job j)
{
    if (job_list_any_p(j)) return; /* already in a linked list */

    j->prev = head->prev;
    j->next = head;
    head->prev->next = j;
    head->prev = j;
}

部分模块源码学习

main.c

int
main(int argc, char **argv) {
    int r;
    // 存放任务的链表
    struct job list = {};

    progname = argv[0];
    // 设置使用行缓存，表使用标准输出作为打印目标
    // 详细文档参见：https://linux.die.net/man/3/setlinebuf
    // 意思是，只有在满足一行（换行）时才输出
    setlinebuf(stdout);
    // 命令行处理
    optparse(&srv, argv + 1);

    if (verbose) {
        printf("pid %d\n", getpid());
    }

    // 服务器端 socket 初始化等，返回一个指向 socket 的 file descriptor
    r = make_server_socket(srv.addr, srv.port);
    if (r == -1) twarnx("make_server_socket()"), exit(111);
    srv.sock.fd = r;

    // 协议处理模块初始化
    prot_init();

    if (srv.user) su(srv.user);
    set_sig_handlers();

    if (srv.wal.use) {
        // We want to make sure that only one beanstalkd tries
        // to use the wal directory at a time. So acquire a lock
        // now and never release it.
        // WAL 即 Write Ahead Log Directory，主要是记录日志用
        // 这里是要保证每次只能有一个 beanstalkd 实例使用 WAL 目录，防止相互写入冲突
        // 那估计以后就没法从 binlog 恢复任务了。。。
        if (!waldirlock(&srv.wal)) {
            twarnx("failed to lock wal dir %s", srv.wal.dir);
            exit(10);
        }

        // 初始化任务链表（双向链表）
        list.prev = list.next = &list;
        // 初始化 WAL，如果 log 中有任务，还要恢复回来，挂载到 job list
        walinit(&srv.wal, &list);
        // 回放任务执行
        r = prot_replay(&srv, &list);
        if (!r) {
            twarnx("failed to replay log");
            return 1;
        }
    }

    // 正式启动 server，并监听请求，处理请求了
    srvserve(&srv);
    return 0;
}

serv.c

void
srvserve(Server *s)
{
    int r;
    Socket *sock;
    int64 period;

    if (sockinit() == -1) {
        twarnx("sockinit");
        exit(1);
    }

    s->sock.x = s;
    // 指定回调，当接收到 `r` 事件后，就会触发这个回调
    s->sock.f = (Handle)srvaccept;
    // Server 维护了关联的客户端连接（最小堆）
    s->conns.less = (Less)connless;
    s->conns.rec = (Record)connrec;

    r = listen(s->sock.fd, 1024);
    if (r == -1) {
        twarn("listen");
        return;
    }

    // 注册
    r = sockwant(&s->sock, 'r');
    if (r == -1) {
        twarn("sockwant");
        exit(2);
    }


    for (;;) {
        // 执行周期性的任务
        // 如果 tick 中执行的任务时间过久，会阻塞后面的 socket connection 处理
        // 严重会导致超时，而如果客户端重试过多，则回增加服务端负载
        period = prottick(s);

        // 轮询是否有就绪的请求（rw），其实就是个适配器，将具体平台下返回的状态
        // 转换成统一的 `r`, `w`, `h`
        // Linux 使用 epoll 封装（参见 `linux.c`）
        // Unix 使用 kqueue 封装（参见 `darwin.c`）
        int rw = socknext(&sock, period);
        if (rw == -1) {
            twarnx("socknext");
            exit(1);
        }

        if (rw) {
            // 如果轮询到需要处理的请求，则执行相应的回调 Handle
            // 注意，这里的回调依然在主线程中执行的，所以如果主线程被阻塞，就呵呵哒了
            sock->f(sock->x, rw);
        }
    }
}

tube.c

// 新建 tube，这里需要给定 tube 名称
// 这里可以看到一个 tube 有几个比较重要的组成：
// 1. 维护就绪任务的堆
// 2. 维护延迟执行任务的堆
// 3. 维护处于 buried 状态的任务链表
// 4. 维护一个等待列表
tube
make_tube(const char *name)
{
    tube t;

    // 分配内存空间，用于存储 tube 结构体值
    t = new(struct tube);
    if (!t) return NULL;

    // 初始化 tube 名称
    t->name[MAX_TUBE_NAME_LEN - 1] = '\0';
    strncpy(t->name, name, MAX_TUBE_NAME_LEN - 1);
    if (t->name[MAX_TUBE_NAME_LEN - 1] != '\0') twarnx("truncating tube name");

    // ready 堆维护着一些已经就绪的 jobs，这里是指定按照 job 优先级的方式比较大小
    // 这样，这个堆顶就是优先级最高的 job
    t->ready.less = job_pri_less;
    // delay 堆维护着一些被延迟执行的 jobs，这里是按照 job delay 时间比较大小
    // 这样，这个堆顶就是延迟时间最短的 job
    t->delay.less = job_delay_less;

    // 用于记录 job 在堆上的位置
    t->ready.rec = job_setheappos;
    t->delay.rec = job_setheappos;
    t->buried = (struct job) { };
    // 使用链表维护着被 bury 掉的 job
    t->buried.prev = t->buried.next = &t->buried;
    // 初始化排队列表
    ms_init(&t->waiting, NULL, NULL);

    return t;
}

// 释放 tube
static void
tube_free(tube t)
{
    // 实际就是从全局的 tubes 列表中移除该 tube
    prot_remove_tube(t);
    // 释放就绪的任务
    free(t->ready.data);
    // 释放延迟的任务
    free(t->delay.data);
    // 清空等待列表
    ms_clear(&t->waiting);
    // 释放 tube 指向的内存
    free(t);
}

// 引用计数：减引用
void
tube_dref(tube t)
{
    if (!t) return;
    if (t->refs < 1) return twarnx("refs is zero for tube: %s", t->name);

    --t->refs;
    // 没有引用后就可以释放该 tube 了
    if (t->refs < 1) tube_free(t);
}

// 引用计数：增加引用
void
tube_iref(tube t)
{
    if (!t) return;
    ++t->refs;
}

// 新建一个 tube，然后注册到全局 tubes 列表
static tube
make_and_insert_tube(const char *name)
{
    int r;
    tube t = NULL;

    t = make_tube(name);
    if (!t) return NULL;

    /* We want this global tube list to behave like "weak" refs, so don't
     * increment the ref count. */
    // 这里是想让全局 tube 列表表现为弱引用，所这里并没有做增加引用的操作
    r = ms_append(&tubes, t);
    // 如果注册 tube 失败，减引用，必要的话会被释放掉
    if (!r) return tube_dref(t), (tube) 0;

    return t;
}

job.c

在 Tube 中有两个最小堆数据结构分别存放被延迟的任何和就绪的任务，这两种使用的排序方式是不同的。我们看到在上面的 Tube 初始化时，给两个堆绑定了不同的比较大小的回调：

t->ready.less = job_pri_less;
t->delay.less = job_delay_less;

以下可以看到具体的排序方式：

// 回调函数，先基于优先级比较哪个小，再基于 id 比较哪个小
int
job_pri_less(void *ax, void *bx)
{
    job a = ax, b = bx;
    if (a->r.pri < b->r.pri) return 1;
    if (a->r.pri > b->r.pri) return 0;
    return a->r.id < b->r.id;
}

// 回调函数，先基于到期时间比较哪个小，再基于 id 比较哪个小
int
job_delay_less(void *ax, void *bx)
{
    job a = ax, b = bx;
    if (a->r.deadline_at < b->r.deadline_at) return 1;
    if (a->r.deadline_at > b->r.deadline_at) return 0;
    return a->r.id < b->r.id;
}

读后感

整个 Beanstalkd 的核心的代码不过五千行左右，但这就实现了一个生产级别的消息队列，的确是很厉害。不过，也正因为其实现比较简单，所以也没有提供类似 Redis 的主从机制等。当然，在这篇文章中，并没有完整地剖析所有的模块实现，只是列出了个人比较感兴趣的模块；关于 binlog 管理的源码（比如垃圾回收，压缩，预留空间申请，任务恢复等）只是粗略地阅读了下，就不在此处献丑啦~

总的来说，对于单机消息队列实现感兴趣的同学还是推荐阅读下该系统的源码，可以学习其中的一些设计思想，实现思路等~

参考

1. beanstalkd 的一些看法
2. beanstalkd repo
3. beanstalkd docs
4. 消息队列 beanstalkd 源码详解
5. 最大—最小堆

延伸阅读

1. Kqueue 与 Epoll 机制
2. 从 0到 100——知乎架构变迁史