twemproxy0.4原理分析-一致性hash演算法實現ketama分析
阿新 • • 發佈：2018-12-13
概述
本文是一致性hash演算法的一種開原始碼的實現：ketama的原始碼分析。
本文是我多年前的一篇文章整理而來，以前的那篇文章的連結可以在這裡檢視。
簡介
若我們在後臺使用NoSQL叢集，必然會涉及到key的分配問題，叢集中某臺機器宕機時如何key又該如何分配的問題。
若我們用一種簡單的方法，n = hash( key)%N來選擇n號伺服器，一切都執行正常，若再考慮如下的兩種情況：
- 一個 cache 伺服器 m down掉了（在實際應用中必須要考慮這種情況），這樣所有對映到 cache m 的物件都會失效，怎麼辦，需要把 cache m 從 cache 中移除，這時候 cache 是 N-1 臺，對映公式變成了 hash(object)%(N-1) ；
- 由於訪問加重，需要新增 cache ，這時候 cache 是 N+1 臺，對映公式變成了 hash(object)%(N+1) ；
1 和 2 意味著什麼？這意味著突然之間幾乎所有的 cache 都失效了。對於伺服器而言，這是一場災難，洪水般的訪問都會直接衝向後臺伺服器； - 再來考慮一個問題，由於硬體能力越來越強，你可能想讓後面新增的節點多做點活，顯然上面的 hash 演算法也做不到。
以上三個問題，可以用一致性hash演算法來解決。關於一致性hash演算法的理論網上很多，這裡分析幾種一致性hash演算法的實現。
ketama實現分析
實現流程介紹
ketama對一致性hash演算法的實現思路是：
- (1) 通過配置檔案，建立一個伺服器列表，其形式如：(1.1.1.1:11211, 2.2.2.2:11211,9.8.7.6:11211…)
- (2) 對每個伺服器列表中的字串，通過Hash演算法，hash成幾個無符號型整數。
注意：如何通過hash演算法來計算呢？ - (3) 把這幾個無符號型整數放到一個環上，這個換被稱為continuum。（我們可以想象，一個從0到2^32的鐘表）
(4) 可以建立一個數據結構，把每個數和伺服器的ip地址對應在一起，這樣，每個伺服器就出現在這個環上的這幾個位置上。
注意：這幾個數，不能隨著伺服器的增加和刪除而變化，這樣才能保證叢集增加/刪除機器後，以前的那些key都對映到同樣的ip地址上。後面將會詳細說明怎麼做。
(5) 為了把一個key對映到一個伺服器上，先要對key做hash，形成一個無符號型整數un，然後在環continuum上查詢大於un的下一個數值。若找到，就把key儲存到這臺伺服器上。
(6) 若你的hash(key)值超過continuum上的最大整數值，就直接回饒到continuum環的開始位置。
這樣，新增或刪除叢集中的結點，就只會影響一少部分key的分佈。
注意：這裡說的會影響一部分key是相對的。其實影響的key的多少，由該ip地址佔的權重大小決定的。在ketama的配置檔案中，需要指定每個ip地址的權重。權重大的在環上佔的點就多。
原始碼分析
在github上下載原始碼後，解壓，進入ketama-master/libketama目錄。一致性hash演算法的實現是在ketama.c檔案中。
在該檔案中，還用到了共享記憶體，這裡不分析這一部分，只分析一致性hash演算法的核心實現部分。
資料結構
// 伺服器資訊，主要記錄伺服器的ip地址和權重值
typedef struct
{
char addr[22]; //伺服器ip地址
unsigned long memory; // 權重值
} serverinfo;
// 以下資料結構就是continuum環上的結點，換上的每個點其實代表了一個ip地址，該結構把點和ip地址一一對應起來。
// 環上的結點
typedef struct
{
unsigned int point; //在環上的點，陣列下標值
char ip[22]; // 對應的ip地址
} mcs;
一致性hash環的建立
該函式是建立continuum的核心函式，它先從配置檔案中讀取叢集伺服器ip和埠，以及權重資訊。建立continuum環，並把這些伺服器資訊和環上的陣列下標對應起來。
// 其中key是為了訪問共享記憶體而設定的，在使用時可以把共享記憶體部分去掉。
static int
ketama_create_continuum( key_t key, char* filename )
{
// 若不使用共享記憶體，可以不管
if (shm_ids == NULL) {
init_shm_id_tracker();
}
// 共享記憶體相關，用不著時，可以去掉
if (shm_data == NULL) {
init_shm_data_tracker();
}
int shmid;
int* data; /* Pointer to shmem location */
// 該變數來記錄共從配置檔案中共讀取了多少個伺服器
unsigned int numservers = 0;
// 該變數是配置檔案中所有伺服器權重值得總和
unsigned long memory;
// 從配置檔案中讀取到的伺服器資訊，包括ip地址，埠，權重值
serverinfo* slist;
// 從配置檔案filename中讀取伺服器資訊，把伺服器總數儲存到變數numservers中，把所有伺服器的權重值儲存到memory中。
slist = read_server_definitions( filename, &numservers, &memory );
/* Check numservers first; if it is zero then there is no error message
* and we need to set one. */
// 以下幾行是檢查讀取的配置檔案內容是否正確
// 若總伺服器數量小於1，錯誤。
if ( numservers < 1 )
{
sprintf( k_error, "No valid server definitions in file %s", filename );
return 0;
}
else if ( slist == 0 ) // 若伺服器資訊陣列為空，錯誤
{
/* read_server_definitions must've set error message. */
return 0;
}
// 以下程式碼開始構建continuum環
/* Continuum will hold one mcs for each point on the circle: */
// 平均每臺伺服器要在這個環上布160個點，這個陣列的元素個數就是伺服器個數*160。
// 具體多少個點，需要根據事情的伺服器權重值進行計算得到。
// 為什麼要選擇160個點呢？主要是通過md5計算出來的是16個整數，把這個整數分成4等分，每份是4位整數。
// 而每進行一次hash計算，我們可以獲得4個點。
mcs continuum[ numservers * 160 ];
unsigned int i, k, cont = 0;
// 遍歷所有伺服器開始在環上部點
for( i = 0; i < numservers; i++ )
{
// 計算伺服器i在所有伺服器權重的佔比
float pct = (float)slist[i].memory / (float)memory;
// 由於計算一次可以得到4個點，所有對每一臺機器來說，總的計算只需要計算40*numservers次。
// 按權重佔比進行劃分，就是以下的計算得到的次數
unsigned int ks = floorf( pct * 40.0 * (float)numservers );
#ifdef DEBUG
int hpct = floorf( pct * 100.0 );
syslog( LOG_INFO, "Server no. %d: %s (mem: %lu = %u%% or %d of %d)\n",
i, slist[i].addr, slist[i].memory, hpct, ks, numservers * 40 );
#endif
// 計算出總次數，每次可以得到4個點
for( k = 0; k < ks; k++ )
{
/* 40 hashes, 4 numbers per hash = 160 points per server */
char ss[30];
unsigned char digest[16];
// 通過計算hash值來得到下標值，該hash值是字串："<ip>-n"，其中的n是通過權重計算出來的該主機應該部點的總數/4。
sprintf( ss, "%s-%d", slist[i].addr, k );
// 計算其字串的md5值，該值計算出來後是一個unsigned char [16]的陣列，也就是可以儲存16個位元組
ketama_md5_digest( ss, digest );
/* Use successive 4-bytes from hash as numbers for the points on the circle: */
// 通過對16個位元組的每組4個位元組進行移位，得到一個0到2^32之間的整數，這樣環上的一個結點就準備好了。
int h;
// 共有16個位元組，可以處理4次，得到4個點的值
for( h = 0; h < 4; h++ )
{
// 把計算出來的連續4位的數字，進行移位。
// 把第一個數字一道一個整數的最高8位，後面的一次移動次高8位，後面一次補零，這樣就得到了一個32位的整數值。移動後
continuum[cont].point = ( digest[3+h*4] << 24 )
| ( digest[2+h*4] << 16 )
| ( digest[1+h*4] << 8 )
| digest[h*4];
// 複製對應的ip地址到該點上
memcpy( continuum[cont].ip, slist[i].addr, 22 );
cont++;
}
}
}
free( slist );
// 以下程式碼對計算出來的環上點的值進行排序，方便進行查詢
// 這裡要注意：排序是按照point的值（計算出來的整數值）進行的，也就是說原來的陣列下標順序被打亂了。
/* Sorts in ascending order of "point" */
qsort( (void*) &continuum, cont, sizeof( mcs ), (compfn)ketama_compare );
// 到這裡演算法的實現就結束了，環上的點(0^32整數範圍內)都已經建立起來，每個點都是0到2^32的一個整數和ip地址的結構。
// 這樣查詢的時候，只是需要hash(key)，並在環上找到對應的數的位置，取得該節點的ip地址即可。
... ...
在環上查詢元素
- 計算key的hash值的實現
unsigned int ketama_hashi( char* inString )
{
unsigned char digest[16];
// 對key的值做md5計算，得到一個有16個元素的unsigned char陣列
ketama_md5_digest( inString, digest );
// 取陣列中的前4個字元，並移位，形成一個整數作為hash得到的值返回
return (unsigned int)(( digest[3] << 24 )
| ( digest[2] << 16 )
| ( digest[1] << 8 )
| digest[0] );
}
- 在環上查詢相應的結點
mcs* ketama_get_server( char* key, ketama_continuum cont )
{
// 計算key的hash值，並儲存到變數h中
unsigned int h = ketama_hashi( key );
// 該變數cont->numpoints是總的陣列埋點數
int highp = cont->numpoints;
// 陣列結點的值
mcs (*mcsarr)[cont->numpoints] = cont->array;
int lowp = 0, midp;
unsigned int midval, midval1;
// divide and conquer array search to find server with next biggest
// point after what this key hashes to
while ( 1 )
{
// 從陣列的中間位置開始找
// 注意此時的陣列是按照point的值排好序了
midp = (int)( ( lowp+highp ) / 2 );
// 若中間位置等於最大點數，直接繞回到0位置
if ( midp == cont->numpoints )
return &( (*mcsarr)[0] ); // if at the end, roll back to zeroth
// 取的中間位置的point值
midval = (*mcsarr)[midp].point;
// 再取一個值：若中間位置下標為0，直接返回0，若中間位置的下標不為0，直接返回上一個結點的point值
midval1 = midp == 0 ? 0 : (*mcsarr)[midp-1].point;
// 把h的值和取的兩個值point值進行比較，若在這兩個point值之間說明h值應該放在較大的那個point值的下標對應的ip地址上
if ( h <= midval && h > midval1 )
return &( (*mcsarr)[midp] );
// 否則繼續2分
if ( midval < h )
lowp = midp + 1;
else
highp = midp - 1;
// 若沒有找到，直接返回0位置的值，這種情況應該很少
if ( lowp > highp )
return &( (*mcsarr)[0] );
}
}
新增刪除機器時會怎樣
先說明一下刪除機器的情況。機器m1被刪除後，以前分配到m1的key需要重新分配，而且最好是均勻分配到現存的機器上。
我們來看看，ketama是否能夠做到？
當m1機器宕機後，continuum環需要重構，需要把m1的ip對應的點從continuum環中去掉。
我們來回顧一下環的建立過程：
按每個ip平均160個點，可以計算出總數t。按每個ip的權重值佔比和總數t的乘積得到該ip應該在該環上部的點數。若一臺機器宕機，那麼每臺機器的權重佔比增加，在該環上部的點數也就相應的增加，當然這個增加也是按每臺機器的佔比來的，佔比多的增加的點數就多，佔比少的增加的點數就少。但，每個ip的點數一定是增加的。
建立環上的點值的過程是：
- 先計算hash值：
for( k = 0; k < ks; k++ ) { //其中ks是每個ip地址對應的總點數
...
sprintf( ss, "%s-%d", slist[i].addr, k );
ketama_md5_digest( ss, digest );
...
}
- 先計算hash值：
continuum[cont].point = ( digest[3+h*4] << 24 )
| ( digest[2+h*4] << 16 )
| ( digest[1+h*4] << 8 )
| digest[h*4];
- 由於此時每個ip的佔比增加，ks就增加了：
// 此時這個值增加
float pct = (float)slist[i].memory / (float)memory;
//該值也增加
unsigned int ks = floorf( pct * 40.0 * (float)numservers );
這樣，每個ip地址對應的point值就多了，但以前的point值不會變。依然在這個環上相同的點值上。也就是說把影響平均分攤到現有的各臺機器上。
當然，刪除的情況和新增的情況相似，都是把影響平均分攤到現有的各個機器上了。
總結
(1) 環上的點是通過對ip地址加一個整數（形如：-N）作為一個字串做hash，然後移位得到4個點數。
(2) 排序後，通過2分查詢進行查詢，效率較高。
(3) 這樣，新增ip時，環上以前部的點不會變化，而且把影響分攤到現有的各個ip上。