gearman（異步計算）學習

 

Gearman是什么？

 

它是分布式的程序調用框架，可完成跨語言的相互調 用，適合在后台運行工作任務。最初是2005年perl版本，2008年發布C/C++版本。目前大部分源碼都是（Gearmand服務job Server）C++，各個API實現有各種語言的版本。PHP的Client API與Worker API實現為C擴展，在PHP官方網站有此擴展的中英文文檔。

 

2Gearman架構中的三個角色

client：請求的發起者，工作任務的需求方（可以是C、PHP、Java、Perl、Mysql udf等等）

Job Server：請求的調度者，負責將client的請求轉發給相應的worker（gearmand服務進程創建）

worker：請求的處理者（可以是C、PHP、Java、Perl等等）

Gearman是如何工作的？

 

從上圖可以看出，Gearman Client API，Gearman Worker API，Gearman Job Server都是由gearman本身提供，我們在應用中只需要調用即可。目前client與worker api都很豐富。

3Gearman的吞吐能力

經過的測試，結果如下：

系統環境：ubuntu-14.0.4 1個CPU 4核 2G內存 （虛擬機）

默認啟動：./gearmand -d

client.php

1. <?php
2. echo "starting...", microtime(true), "n";
3. $gmc = new GearmanClient();
4. $gmc->setCompleteCallBack(function($task){
5. //echo $task->data(), "n";
6. });
7. $gmc->addServer("127.0.0.1", 4730);
8. for ($i = 0; $i < 100000; $i++) {
9. $gmc->addTaskBackground("reserve", "just test it", null, $i);
10. }
11. $gmc->runTasks();
12. echo "end...", microtime(true), "n";

worker.php

1. <?php
2. $gmw = new GearmanWorker();
3. $gmw->addServer("127.0.0.1", 4730);
4. $gmw->addFunction("reserve", function($job) {
5. if ($job->unique() == 99999) {
6. echo microtime(true), "n";
7. }
8. return strrev($job->workload());
9. });
10. while($gmw->work());

啟動一個job server實例：job server IP:127.0.0.1 PORT:4730

 

啟動一個worker: php worker.php

worker注冊reserve函數，將client的job字符串反轉后返回。

client工作任務的消息為：just test it（12字節）

同步：4100/s

異步：25700/s

memcached內存准持久化的吞吐能力測試

./gearmand -d -q libmemcached —libmemcached-servers=127.0.0.1:11211

client投遞100000個工作任務：16400/s

Gearman典型的部署結構

 

Gearman支持的特性

高可用

啟動兩個job server，他們是獨立的服務進程，有各自的內存隊列。當一個job server進程出現故障，另一個job server可以正常調度。（worker api與client api可以完成job server故障的切換）。在任何時候我們可以關閉某個worker，即使那個worker正在處理工作任務（Gearman不會讓正在被執行的job丟失的，由於worker在工作時與Job server是長連接，所以一旦worker發生異常，Job server能夠迅速感知並重新派發這個異常worker剛才正在執行的工作）

負載均衡（附gearman協議會詳細解釋）

job server並不主動分派工作任務，而是由worker從空閑狀態喚醒之后到job server主動抓取工作任務。

可擴展

松耦合的接口和無狀態的job，只需要啟動一個worker，注冊到Job server集群即可。新加入的worker不會對現有系統有任何的影響。

分布式

gearman是分布式的任務分發框架，worker與job server，client與job server通信基於tcp的socket連接。

隊列機制

gearman內置內存隊列，默認情況隊列最大容量為300W，可以配置最大支持2^32-1，即4 294 967 295。

高性能

作為Gearman的核心，Job server的是用C/C++實現的，由於只是做簡單的任務派發，因此系統的瓶頸不會出在Job server上。

兩種工作任務

后台工作任務Background job——時序圖

 

由圖可知，client提交完job，job server成功接收后返回JOB_CREATED響應之后，client就斷開與job server之間的鏈接了。后續無論發生什么事情，client都是不關心的。同樣，job的執行結果client端也沒辦法通過Gearman消息框架 獲得。

一般工作任務Non-background job——時序圖

由圖可知，client端在job執行的整個過程中，與job server端的鏈接都是保持着的，這也給job完成后job server返回執行結果給client提供了通路。同時，在job執行過程當中，client端還可以發起job status的查詢。當然，這需要worker端的支持的。

4關於持久化

對於隊列持久化的問題，是一個值得考慮的問題。持久化必然影響高性能。gearman支持后台工作任務的持久化，支持drizzle、mysql、memcached的持久化。對於client提交的background job，Job server除了將其放在內存隊列中進行派發之外，還會將其持久化到外部的持久化隊列中。一旦Job server發生問題重啟，外部持久化隊列中的background job將會被恢復到內存中，參與Job server新的派發當中。這保證了已提交未執行的background job不會由於Job server發生異常而丟失。並且我測試發現如果開啟了持久化，那么后台工作任務會先將工作任務寫到持久化介質，然后在入內存隊列，再執行。非后台工作任務，由於client與job server是保持長連接的狀態，如果工作任務執行異常，client可以靈活處理，所以無須持久化。

Gearman框架中的一個問題

從典型部署結構看出，兩個Job server之間是沒有連接的。也就是Job server間是不共享background job的。如果通過讓兩個Job server指向同一個持久化隊列，可以讓兩個Job serer互相備份。但實際上，這樣是行不通的。因為Job server只有在啟動時才會將持久化隊列中的background job轉入到內存隊列。也就是說，Job server1如果宕機且永遠不啟動，Job server2一直正常運行，那么Job server1宕機前被提交到Job server1的未被執行的background job將永遠都呆在持久化隊列中，得不到執行。另外如果多個job server實例指向同一個持久化隊列，同時重啟多個job server實例會導致持久隊列中的工作任務被多次載入，從而導致消息重復處理。

我建議的部署結構

 

采用memcached做后台工作任務的准持久化隊列，最好memcached和job server在內網的不同機器。兩個機器的兩個服務同時掛掉的可能性比較小，同時也保證了高性能。而且memcached應該為兩個相互獨立實例，防止其上述的gearman框架中的問題。我們可以做一個監控腳本，如果某個job server異常退出，可以重啟，也最大化的保證了job server的高可用。

5關於Gearman的管理工具

目前有一個現在的管理工具，https://github.com/brianlmoon/GearmanManager，但是只支持php-5.2，不過可以自行修改支持php-5.4，我建議如果使用PHP作為worker進程，使用php-5.4以上的版本。該工具的設計方法可以借鑒，可以比較好的管理gearman worker。

應用場景

1. 結合linux crontab，php腳本負責產生job，將任務分發到多台服務器周期性的並發執行。可以取代目前我們比較多的crontab的工作任務。

2. 郵件短信發送

3. 異步log

4. 跨語言相互調用（對於密集型計算的需求，可以用C實現，PHP直接調用）

5. 其他耗時腳本

Gearman安裝（unbuntu）

下載

 

$>wget https://launchpadlibrarian.net/165674261/gearmand-1.1.12.tar.gz 

安裝依賴包

$>sudo apt-get install libboost1.55-all-dev gperf libevent libevent-dev uuid libmemcached-dev  
$>tar zxvf gearmand-1.1.12.tar.gz 
$>cd gearmand-1.1.12 
$>/configure --prefix=/home/phpboy/Server/gearman 
$>make & make install 

啟動

a)默認啟動

 

$>./gearman -d 

b)支持memcached准持久化

$>./gearmand -d -q libmemcached --libmemcached-servers=127.0.0.1:11211 

安裝php的Gearman擴展

$>wget http://pecl.php.net/get/gearman-1.1.2.tgz 
$>tar zxvf gearman-1.1.2.tgz#cd gearman-1.1.2 
$>phpize 
$>./configure --with-php-config=php-config 
$>make & make install 

php client api與php worker api

可以用上面我的測試的示例

附Gearmand（job server的啟動參數簡單說明）

-b, –backlog=BACKLOG 連接請求隊列的最大值 
-d, –daemon Daemon 守護進程化  
-f, –file-descriptors=FDS 可打開的文件描述符數量  
-h, –help  
-l, –log-file=FILE Log 日志文件 
-L, –listen=ADDRESS 開啟監聽的地址  
-p, –port=PORT 開啟監聽的端口  
-P, –pid-file=FILE File pid file  
-r,–protocol=PROTOCOL 使用的協議  
-q, –queue-type=QUEUE 持久化隊列類型  
-t, –threads=THREADS I/O線程數量  
-u, –user=USER 進程的有效用戶名  
libdrizzle Options:  
--libdrizzle-host=HOST Host of server.   
--libdrizzle-port=PORT Port of server.   
--libdrizzle-uds=UDS Unix domain socket for server.   
--libdrizzle-user=USER User name for authentication.   
--libdrizzle-password=PASSWORD Password for authentication.   
--libdrizzle-db=DB Database to use.   
--libdrizzle-table=TABLE Table to use.   
--libdrizzle-mysql Use MySQL protocol.  
libmemcached Options:   
--libmemcached-servers=SERVER_LIST List of Memcached servers to use.  
libsqlite3 Options:   
--libsqlite3-db=DB Database file to use.   
--libsqlite3-table=TABLE Table to use.  
libpq Options:   
--libpq-conninfo=STRING PostgreSQL connection information string.   
--libpq-table=TABLE Table to use.  
http Options:   
--http-port=PORT Port to listen on. 

Gearman通信協議

總括

Gearman工作在TCP上，默認端口為4730，client與job server、worker與job server的通信都基於此tcp的socket連接。client是工作任務的發起者，worker是可以注冊處理函數的工作任務執行者，job server為工作的調度者。協議包含請求報文與響應報文兩個部分，所有發向job server的數據包（TCP報文段的數據部分）認為是請求報文，所有從job server發出的數據包（TCP報文段的數據部分）認為是響應報文。worker或者client與job server間的通信是基於二進制數據流的，但在管理client也有基於行文本協議的通信。

請求的報文體

 

響應的報文體

后台工作任務Background job

一般工作任務Non-background job

二進制包

請求報文與響應報文是由二進制包封裝。一個二進制包由頭header和可選的數據部分data組成。

header的組成

1. 報文類別，請求報文或者響應報文，4個字節
   “�REQ” 請求報文
   “�RES” 響應報文

2. 包類型，高（大）字節序（網絡字節序），4個字節可能的類型有

 

類型值  名稱    報文類型 發送者   

 

 

1 CAN_DO REQ    Worker    
2 CANT_DO REQ    Worker    
3 RESET_ABILITIES     REQ    Worker    
4 PRE_SLEEP           REQ    Worker    
5 (unused)            -      -   
6 NOOP                RES    Worker    
7 SUBMIT_JOB          REQ    Client    
8 JOB_CREATED         RES    Client    
9 GRAB_JOB            REQ    Worker    
10 NO_JOB              RES    Worker    
11 JOB_ASSIGN          RES    Worker    
12 WORK_STATUS         REQ    Worker    
13 WORK_COMPLETE REQ    Worker   
14 WORK_FAIL           REQ    Worker  
15 GET_STATUS          REQ    Client  
16 ECHO_REQ            REQ    Client/Worker  
17 ECHO_RES            RES    Client/Worker  
18 SUBMIT_JOB_BG       REQ    Client 
19 ERROR               RES    Client/Worker  
20 STATUS_RES          RES    Client  
21 SUBMIT_JOB_HIGH     REQ    Client  
22 SET_CLIENT_ID       REQ    Worker  
23 CAN_DO_TIMEOUT REQ    Worker  
24 ALL_YOURS           REQ    Worker  
25 WORK_EXCEPTION      REQ    Worker  
26 OPTION_REQ          REQ    Client/Worker  
27 OPTION_RES          RES    Client/Worker  
28 WORK_DATA           REQ    Worker    
29 WORK_WARNING        REQ    Worker   
30 GRAB_JOB_UNIQ       REQ    Worker  
31 JOB_ASSIGN_UNIQ     RES    Worker  
32 SUBMIT_JOB_HIGH_BG  REQ    Client  
33 SUBMIT_JOB_LOW      REQ    Client  
34 SUBMIT_JOB_LOW_BG   REQ    Client  
35 SUBMIT_JOB_SCHED    REQ    Client  
36 SUBMIT_JOB_EPOCH    REQ    Client 

 

1. 可選數據部分長度，高（大）字節序（網絡字節序），4個字節，可表示的值為4294967295

數據部分，數據部分的各個部分為null字符分隔。

具體各包類型的說明

client和worker可發送請求報文

ECHO_REQ

當job server收到此包類型的請求報文時，就簡單的產生一個包類型為ECHO_RES，同時將請求報文的數據部分作為響應報文的數據部分的報文。主要用於測試或者調試

如：

Client -> Job Server  00 52 45 51 0REQ  報文類型  00 00 00 a0 16 (Packet type: ECHO_ERQ)    00 00 00 04 4 (Packet length)  74 65 73 74 test   (Workload) 

ECHO_RESclient和worker可接收響應報文

當job server響應ECHO_REQ報文時發送的包類型為ECHO_RES的響應報文

如：

Job Server -> Client  00 52 45 53 0RES  報文類型  00 00 00 a1 17 (Packet type: ECHO_ERS)  00 00 00 04 4 (Packet length)  74 65 73 74 test   (Workload) 

當job server發生錯誤時，需要通知client或者workerERROR

client發送的請求報文：（僅能由client發送的請求報文）
SUBMIT_JOB, SUBMIT_JOB_BG,SUBMIT_JOB_HIGH, SUBMIT_JOB_HIGH_BG,SUBMIT_JOB_LOW, SUBMIT_JOB_LOW_BG

當client有一個工作任務需要運行，就會提交相應的請求報文，job server響應包類型為JOB_CREATED數據部分為job handle的響應報文。SUBMIT_JOB為普通的工作任務，client得到狀態更新及通知任務已經完成的響應；SUBMIT_JOB_BG為異步的工作任務，client不關心任務的完成情況；SUBMIT_JOB_HIGH為高優先級的工作任務，SUBMIT_JOB_HIGH_BG為高優先級的異步任務；SUBMIT_JOB_LOW為低優先級的工作任務，SUBMIT_JOB_LOW_BG為低優先級的異步任務。

如：

1 2 3 4 5 6 7 8 9 10 11 12 13

Client -> Job Server 00 52 45 51 0REQ (報文類型) 00 00 00 07 7 (Packet type: SUBMIT_JOB) 00 00 00 0d 13 (Packet length) 72 65 76 65 72 73 65 00 reverse0 (Function) 00 � (Unique ID) 74 65 73 74 test (Workload)

SUBMIT_JOB_SCHED

和SUBMIT_JOB_BG類似，此類型的工作任務不會立即執行，而在設置的某個時間運行。

如：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

Client -> Job Server 00 52 45 51 0REQ (報文類型) 00 00 00 23 35 (Packet type: SUBMIT_JOB_SCHED) 00 00 00 0d 13 (Packet length) 72 65 76 65 72 73 65 00 reverse0 (Function) 00 � (Unique ID) 01 � (minute 0-59) 01 � (hour 0-23) 01 � (day of month 1-31) 01 � (day of month 1-12) 01 � (day of week 0-6) 74 65 73 74 test (Workload)

SUBMIT_JOB_EPOCH

和SUBMIT_JOB_SCHED作用一樣，只是將設置的時間定為了uinx時間戳GET_STATUS獲取某個工作任務執行的狀態信息

OPTION_REQ

設置client與job server連接的選項

client獲取的響應報文：

JOB_CREATED響應包類型為SUBMIT_JOB*的請求報文，數據部分為job handle

WORK_DATA, WORK_WARNING, WORK_STATUS, WORK_COMPLETE,WORK_FAIL, WORK_EXCEPTION

對於后台運行的工作任務，任務執行信息可以通過包類型為上面的值來查看。

STATUS_RES

響應包類型為GET_STATUS的請求報文，通常用來查看一個后台工作任務是否已經完成，以及完成的百分比。

OPTION_RES

響應包類型為OPTION_REQ的請求報文

worker發送的請求報文：

CAN_DO

通知job server可以執行給定的function name的任務，此worker將會放到一個鏈表，當job server收到一個function name的工作任務時，worker為被喚醒。

CAN_DO_TIMEOUT

和CAN_DO類似，只是針對給定的function_name的任務設置了一個超時時間。

CANT_DO

worker通知job server已經不能執行給定的function name的任務

RESET_ABILITIES

worker通知job server不能執行任何function name的任務

PRE_SLEEP

worker通知job server它將進入sleep階段，而之后此worker會被包類型為NOOP的響應報文喚醒。

GRAB_JOB

worker向job server抓取工作任務，job server將會響應NO_JOB或者JOB_ASSIG

NGRAB_JOB_UNIQ

和GRAB_JOB類似，但是job server在有工作任務時將會響應JOB_ASSIGN_UNIQ

WORK_DATA

worker請求報文的數據部分更新client

WORK_WARNING

worker請求報文代表一個warning，它應該被對待為一個WARNING

WORK_STATU

Sworker更新某個job handle的工作狀態，job server應該儲存這些信息，以便響應之后client的GET_STATUS請求

WORK_COMPLETE

通知job server及所有連接的client，數據部分為返回給client的數據

WORK_FAIL

通知job server及所有連接的client，工作任務執行失敗

WORK_EXCEPTION

通知job server及所有連接的client，工作任務執行失敗並給出相應的異常

SET_CLIENT_ID

設置worker ID，從而job server的控制台及報告命令可以標識各個worker，數據部分為worker實例的標識

ALL_YOURS

暫未實現

worker獲取的響應報文：

NOOP

job server喚醒sleep的worker，以便可以開始抓取工作任務

NO_JOB

job server響應GRAB_JOB的請求，通知worker沒有等待執行的工作任務

JOB_ASSIGN

job server響應GRAB_JOB的請求，通知worker有需要執行的工作任務

JOB_ASSIGN_UNIQ

job server響應GRAB_JOB_UNIQ的請求，和JOB_ASSIGN一樣，只是為client傳遞了一個唯一標識

基於上述的協議描述一個完整的例子

worker注冊可以執行的工作任務

Worker -> Job Server

1 2 3 4 5 6 7

00 52 45 51 0REQ (Magic) 00 00 00 01 1 (Packet type: CAN_DO) 00 00 00 07 7 (Packet length) 72 65 76 65 72 73 65 reverse (Function)

worker檢測或者抓取工作任務

1 2 3 4 5 6 7

Worker -> Job Server 00 52 45 51 0REQ (Magic) 00 00 00 09 9 (Packet type: GRAB_JOB) 00 00 00 00 0 (Packet length)

job server響應worker的抓取工作（沒有工作任務）

1 2 3 4 5

00 52 45 53 0RES (Magic) 00 00 00 0a 10 (Packet type: NO_JOB) 00 00 00 00 0 (Packet length)

worker通知job server開始sleep

1 2 3 4 5

00 52 45 51 0REQ (Magic) 00 00 00 04 4 (Packet type: PRE_SLEEP) 00 00 00 00 0 (Packet length)

client提交工作任務

1 2 3 4 5 6 7 8 9 10 11 12 13

Client -> Job Server 00 52 45 51 0REQ (Magic) 00 00 00 07 7 (Packet type: SUBMIT_JOB) 00 00 00 0d 13 (Packet length) 72 65 76 65 72 73 65 00 reverse0 (Function) 00 � (Unique ID) 74 65 73 74 test (Workload)

job server響應client的SUBMIT_JOB請求，返回job handle

1 2 3 4 5 6 7

00 52 45 53 0RES (Magic) 00 00 00 08 8 (Packet type: JOB_CREATED) 00 00 00 07 7 (Packet length) 48 3a 6c 61 70 3a 31 H:lap:1 (Job handle)

job server喚醒worker

1 2 3 4 5 6 7

Job Server -> Worker 00 52 45 53 0RES (Magic) 00 00 00 06 6 (Packet type: NOOP) 00 00 00 00 0 (Packet length)

worker的抓取工作任務

1. Worker -> Job Server
2.  
3. 00 52 45 51 \0REQ (Magic)
4.  
5. 00 00 00 09 9 (Packet type: GRAB_JOB)
6.  
7. 00 00 00 00 0 (Packet length)

job server分配工作任務給worker

1. Job Server -> Worker
2.  
3. 00 52 45 53 \0RES (Magic)
4.  
5. 00 00 00 0b 11 (Packet type: JOB_ASSIGN)
6.  
7. 00 00 00 14 20 (Packet length)
8.  
9. 48 3a 6c 61 70 3a 31 00 H:lap:1\0 (Job handle)
10.  
11. 72 65 76 65 72 73 65 00 reverse\0 (Function)
12.  
13. 74 65 73 74 test (Workload)

worker完成工作任務通知job server

1. 00 52 45 51 \0REQ (Magic)
2.  
3. 00 00 00 0d 13 (Packet type: WORK_COMPLETE)
4.  
5. 00 00 00 0c 12 (Packet length)
6.  
7. 48 3a 6c 61 70 3a 31 00 H:lap:1\0 (Job handle)
8.  
9. 74 73 65 74 tset (Response)

job server通知client完成了工作任務

 

Job Server -> Client

1. 00 52 45 53 \0RES (Magic)
2.  
3. 00 00 00 0d 13 (Packet type: WORK_COMPLETE)
4.  
5. 00 00 00 0c 12 (Packet length)
6.  
7. 48 3a 6c 61 70 3a 31 00 H:lap:1\0 (Job handle)
8.  
9. 74 73 65 74 tset (Response)

每個client與job server是全雙工通信，在一個socket可以完成多個工作任務的投遞，但是收到任務的執行結果的順序可能與投遞的順序不一致。

 

6總結worker的工作流程

1. Worker通過CAN_DO消息，注冊到Job server上。

2. 隨后發起GRAB_JOB，主動要求分派任務。

3. Job server如果沒有job可分配，就返回NO_JOB。

4. Worker收到NO_JOB后，進入空閑狀態，並給Job server返回PRE_SLEEP消息，告訴Job server:”如果有工作來的話，用NOOP請求我先。”

5. Job server收到worker的PRE_SLEEP消息后，明白了發送這條消息的worker已經進入了空閑態。

6. 這時如果有job提交上來，Job server會給worker先發一個NOOP消息。

7. Worker收到NOOP消息后，發送GRAB_JOB向Job server請求任務。

8. Job server把工作派發給worker。

9. Worker干活，完事后返回WORK_COMPLETE給Job server。