高效能分散式佇列系統 Beanstalkd 介紹及使用

語言: CN / TW / HK

時間 2021-07-29 21:01:15 FalconChen

主題: fsync node.js github javascript beanstalkd freebsd linux rhel centos mac os x

Beanstalkd是一個簡單、高效的工作佇列系統，其最初設計目的是通過後臺非同步執行耗時任務方式降低高容量Web應用的頁面延時。而其簡單、輕量、易用等特點，和對任務優先順序、延時超時重發等控制，以及眾多語言版本的客戶端的良好支援，使其可以很好的在各種需要佇列系統的場景中應用。

1. `Beanstalkd`介紹

1.1 核心概念

Beanstalkd 架構

Beanstalkd使用Producer-Consumer設計模式，無論是其協議結構還是使用方式都是類Memcached風格的。以下是Beanstalkd設計思想中核心概念：

job - 任務

job是一個需要非同步處理的任務，是Beanstalkd中的基本單元，job需要放在一個tube中。

Beanstalkd中的任務（job）類似於其它佇列系統中的訊息（message）的概念，詳細參考任務生命週期

tube - 管道

管道即某一種型別的任務佇列，其類似與訊息的主題（topic）,是Producer和Consumer的操作物件。

一個Beanstalkd中可以有多個管道, 每個管道都有自己的釋出者（Producer）和消費者Consumer，管道之間互相不影響。

producer - 生產者

任務（job）的生產者，通過put命令來將一個job放到一個tube中。

consumer - 消費者

任務（job）的消費者，通過reserve、release、bury、delete命令來獲取或改變job的狀態。

1.2 任務生命週期

Beanstalkd中的任務（job）替代了訊息（message）的概念，任務會有一系列的狀態。任務的生命週期如下：

Beanstalkd-job-status

一個Beanstalkd任務可能會包含以下狀態：

READY - 需要立即處理的任務。當producer直接put一個任務時，任務就處於READY狀態，以等待consumer來處理。當延時 (DELAYED) 任務到期後會自動成為當前READY狀態的任務
DELAYED - 延遲執行的任務。當任務被延時put時，任務就處於DELAYED狀態。等待時間過後，任務會被遷移到READY狀態。當消費者處理任務後，可以用將訊息再次放回DELAYED佇列延遲執行
RESERVED - 已經被消費者獲取，正在執行的任務。當consumer獲取了當前READY的任務後，該任務的狀態就會遷移到RESERVED狀態，這時其它的consumer就不能再操作該任務。Beanstalkd會檢查任務是否在TTR(time-to-run)內完成
BURIED - 保留的任務，這時任務不會被執行，也不會消失。當consumer完成該任務後，可以選擇delete、release或者bury操作。
delete後，任務會被刪除，生命週期結束；

release操作可以重新把任務狀態遷移回READY狀態或DELAYED狀態，使其他consumer可以繼續獲取和執行該任務

bury會拔任務休眠，等需要該任務時，再將休眠的任務kick回READY；也可能過delete刪除BURIED狀態的任務
DELETED - 訊息被刪除，Beanstalkd不再維持這些訊息。即任務生命週期結束。

任務優先順序（priority）

任務 (job) 可以有0~2^32個優先順序，0表示優先順序最高。Beanstalkd採用最大最小堆 (Min-max heap) 處理任務優先順序排序, 任何時刻呼叫 reserve 命令的消費者總是能拿到當前優先順序最高的任務, 時間複雜度為 O(logn)

任務延時（delay）

Beanstalkd中可以通過兩種方式延時執行任務：生產者釋出任務時指定延時；或者當任務處理完畢後, 消費者再次將任務放入佇列延時執行 (RELEASE with delay)。這種機制可以實現分散式定時任務，這種任務機制的優勢是：如果某個消費者節點故障，任務超時重發（time-to-run）以保證任務轉移到其它節點執行。

任務超時重發（time-to-run）

Beanstalkd把任務返回給消費者後：消費者必須在預設的TTR(time-to-run) 時間內傳送delete、release或者bury命令改變任務狀態；否則Beanstalkd會認為訊息處理失敗，然後把任務交給另外的消費者節點執行。如果消費者預計在TTR時間內無法完成任務, 可以傳送touch命令，以使Beanstalkd重新計算TTR

任務預留（buried）

當RESERVED狀態的任務因為某些原因無法執行時，消費者可以將其設定為buried狀態，這時Beanstalkd會繼續保留這些任務。在具備任務執行條件時，再通過kick將任務遷移回READY狀態。

2. `Beanstalkd`安裝使用

Beanstalkd分為服務端和客戶端兩部分。可以在其官網查詢相關安裝包及安裝方法：

服務端：http://kr.github.io/beanstalkd/download.html
客戶端：https://github.com/kr/beanstalkd/wiki/client-libraries

2.1 服務端

要使用Beanstalkd，首先需要在一或多臺機器安裝並執行beanstalkd服務端。安裝beanstalkd需要Linux (2.6.17 or later) 、Mac OS X、或 FreeBSD，可以通過原始碼編譯或安裝包來安裝。

原始碼安裝

下載、解壓並進入原始碼目錄後，執行make或make install命令即可：

$ sudo make
// 或
$ sudo make install 
// 或
$ sudo make install PERFIX=/usr/bin/beanstalkd

安裝包安裝

在Unbuntu或Debian系統中，可以使用以下命令安裝：

$ sudo apt-get install beanstalkd

在CentOS或RHEL系統中，首先需要更新EPEL源，然後再使用yum命令安裝。

在RHEL6中使用以下命令更新源：

su -c 'rpm -Uvh http://download.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm'

在RHEL7中：

su -c 'rpm -Uvh http://download.fedoraproject.org/pub/epel/7/x86_64/e/epel-release-7-9.noarch.rpm'

執行安裝：

$ sudo yum install beanstalkd

執行beanstalkd

前臺執行：

$ beanstalkd

這時，beanstalkd會保持在前臺，可以通過control+C命令結束程序。

要使beanstalkd後臺執行，可以在命令結尾增加&：

$ beanstalkd &

啟動beanstalkd時還可以增加一些啟動選項：

$ beanstalkd -l 127.0.0.1 -p 11300 &

在以上命令中，我們通過-l指定了監聽地址、-p引數指定了監聽埠、&指定為後臺執行。

beanstalkd執行引數

Beanstalkd安裝後，就可以通過beanstalkd命令來啟動或配置Beanstalkd。該命令的使用格式如下：

beanstalkd [OPTIONS]

可選[OPTIONS]引數有：

-b DIR - wal目錄
-f MS - 指定MS毫秒內的 fsync (-f0 為"always fsync")
-F - 從不 fsync (預設)
-l ADDR - 指定監聽地址（預設為：0.0.0.0）
-p PORT - 指定監聽埠（預設為：11300）
-u USER - 使用者與使用者組
-z BYTE - 最大的任務大小（預設為：65535）
-s BYTE - 每個wal檔案的大小（預設為：10485760）
-c - 壓縮binlog（預設）
-n - 不壓縮binlog

2.2 客戶端

客戶端包含了Beanstalkd設計概念中的任務生產者（Producer）和消費者（Consumer）。Beanstalkd有很多語言版本客戶端的實現，點選Beanstalkd 客戶端查詢自大所需要的版本，如果都不能滿足需要，還可以根據Beanstalkd 協議自行實現。

筆者日常工作中，接觸Node.js語言較多，以下用一個Node.js版本的Beanstalkd 客戶端：fivebeans為例，簡單演示Beanstalkd的任務處理流程。

安裝fivebeans模組後，建立一個consumer客戶端。程式碼如下：

var fivebeans = require('fivebeans');

var consumer = new fivebeans.client('192.168.3.218', 11300);

// 連線服務端
consumer.connect();
// 臨聽名為 itbilu 的tube
consumer.watch('itbilu', function(err, numwatched) {})
// 收到任務後，處理任務
consumer.reserve(function(err, jobid, payload) {
  console.log('收到任務：%s，任務內容：%s', jobid, payload);
});

再建立一個producer客戶端。程式碼如下：

var fivebeans = require('fivebeans');

var producer = new fivebeans.client('192.168.3.218', 11300);
// 連線服務端
producer.connect();
// 使用名為 itbilu 的tube
producer.use('itbilu', function(err, tubename) {});
// 向tube 釋出任務
producer.put(1024, 0, 2, '內容', function(err, jobid) {
  console.log('釋出了一個任務，任務ID：%s', jobid);
});

完成並儲存程式碼後，首先執行客戶端consumer.js：

node consumer.js

然後執行客戶端producer.js：

node producer.js

producer執行後，會自動put一個任務，而consumer處於監聽狀態，就會收到這個任務：

// procuer 執行後
釋出了一個任務，任務ID：1
// consumer 收到任務後
收到任務：1，任務內容：內容

「其他文章」