こんにちは！分散システムの重要な技術である「一貫性のあるハッシュ（Consistent Hashing）」について解説した記事を紹介します。
はじめに：なぜConsistent Hashingが必要なのか？
現代のWebサービスでは、大量のデータを複数のサーバーに分散して保存する必要があります。たとえば、ユーザーがアップロードしたファイルや、キャッシュデータなどをどのサーバーに保存するか決める必要があります。
通常のハッシュ関数を使う場合、以下のように実装するかもしれません：
def hash_fn(key):
"""キーをバイト合計し、サーバー数（5台）で割ったあまりを返す"""
return sum(bytearray(key.encode('utf-8'))) % 5
しかし、この方法には大きな問題があります。サーバーの台数が変わったとき、多くのデータの再配置が必要になるのです。
例えば、5台のサーバーがあるとします。ファイル「file1.txt」は hash_fn("file1.txt") = 2 なので、3番目のサーバーに保存されます。しかし、1台サーバーを追加して6台になると、ハッシュ関数は % 6 に変わり、多くのファイルの配置先が変わってしまいます。
これは大規模なデータ移動を引き起こし、システムに大きな負荷をかけます。この問題を解決するのが「Consistent Hashing」なのです。
Consistent Hashingの基本概念
Consistent Hashingの基本的なアイデアは以下の通りです：
- 巨大で固定サイズのハッシュ空間（たとえば0〜2^256-1）を考える
- サーバーもデータも同じハッシュ空間上にマッピングする
- データは「右回りで最も近いサーバー」に割り当てる
これにより、サーバーが追加・削除されても影響を受けるのは、そのサーバーの近くにあるデータだけになります。
ハッシュリングを理解する
Consistent Hashingは「ハッシュリング」と呼ばれる円環状の構造で説明されることが多いです。
このリング上に、サーバーとデータの両方がマッピングされます。データはリング上で時計回りに進み、最初に見つかるサーバーに割り当てられます。
Consistent Hashingの実装方法
実装方法としては、以下のアプローチが考えられます：
- 巨大な配列を使う方法（メモリ効率が悪い）
- 二分探索を使った効率的な方法（実用的）
今回は2番目の方法で実装してみましょう。Pythonのコードで説明します。
基本構造
import hashlib
import bisect
class ConsistentHash:
def __init__(self, total_slots=2**256):
"""
ConsistentHashを初期化
total_slots: ハッシュ空間のサイズ
"""
self.total_slots = total_slots
self._keys = [] # サーバーのハッシュ値を保存するリスト
self.nodes = [] # サーバーノードを保存するリスト
def hash_fn(self, key):
"""
キーをハッシュ値に変換する関数
SHA-256ハッシュを計算し、total_slotsで割ったあまりを返す
"""
hsh = hashlib.sha256()
hsh.update(bytes(key.encode('utf-8')))
return int(hsh.hexdigest(), 16) % self.total_slots
ノードの追加
def add_node(self, node):
"""
新しいノードをハッシュリングに追加する
node: 追加するStorageNodeオブジェクト
"""
if len(self._keys) == self.total_slots:
raise Exception("ハッシュ空間がいっぱいです")
# ノードの位置をハッシュ関数で計算
key = self.hash_fn(node.host)
# キーを挿入する位置を二分探索で見つける
index = bisect.bisect(self._keys, key)
# 同じキーが既にある場合はエラー
if index > 0 and self._keys[index - 1] == key:
raise Exception("衝突が発生しました")
# この時点でデータ移行が必要になる
# （実際のコードではここでデータ移行処理を行う）
# ノードとキーを挿入
self.nodes.insert(index, node)
self._keys.insert(index, key)
return key
ノードの削除
def remove_node(self, node):
"""
ハッシュリングからノードを削除する
node: 削除するStorageNodeオブジェクト
"""
if len(self._keys) == 0:
raise Exception("ハッシュ空間が空です")
# ノードの位置をハッシュ関数で計算
key = self.hash_fn(node.host)
# キーの位置を二分探索で検索
index = bisect.bisect_left(self._keys, key)
# キーが見つからない場合はエラー
if index >= len(self._keys) or self._keys[index] != key:
raise Exception("ノードが存在しません")
# この時点でデータ移行が必要になる
# （実際のコードではここでデータ移行処理を行う）
# ノードとキーを削除
self._keys.pop(index)
self.nodes.pop(index)
return key
アイテムの割り当て
def assign(self, item):
"""
アイテムを適切なノードに割り当てる
item: 割り当てるアイテムの識別子
戻り値: 割り当てられたノード
"""
if len(self._keys) == 0:
raise Exception("ノードがありません")
# アイテムの位置をハッシュ関数で計算
key = self.hash_fn(item)
# アイテムの右側にある最初のノードのインデックスを見つける
# bisect_rightはキー以上の最小のインデックスを返す
index = bisect.bisect_right(self._keys, key) % len(self._keys)
# そのインデックスにあるノードを返す
return self.nodes[index]
Golangによる実装
Consistent Hashingを実際にGolangで実装してみましょう。以下の実装では、より効率的なmurmur3ハッシュ関数を使用しています。
基本的な構造体
まず、ノードとハッシュリングを表す構造体を定義します：
package main
import (
"errors"
"fmt"
"sort"
"sync"
"github.com/spaolacci/murmur3"
)
// Node represents a server in our distributed system
type Node struct {
ID string
Keys map[string]string
}
// ConsistentHashRing manages the hash ring and node assignments
type ConsistentHashRing struct {
mu sync.RWMutex
nodes map[uint32]*Node
hashes []uint32
}
// NewConsistentHashRing creates a new consistent hash ring
func NewConsistentHashRing() *ConsistentHashRing {
return &ConsistentHashRing{
nodes: make(map[uint32]*Node),
hashes: []uint32{},
}
}
// ハッシュ関数としてmurmur3を使用
func hashFunction(key string) uint32 {
return murmur3.Sum32([]byte(key))
}
ノードの追加
ハッシュリングにノードを追加する関数を実装します：
// AddNode adds a new node to the hash ring
func (chr *ConsistentHashRing) AddNode(id string) {
chr.mu.Lock()
defer chr.mu.Unlock()
hash := hashFunction(id)
chr.nodes[hash] = &Node{
ID: id,
Keys: make(map[string]string),
}
chr.hashes = append(chr.hashes, hash)
sort.Slice(chr.hashes, func(i, j int) bool { return chr.hashes[i] < chr.hashes[j] })
}
キーの割り当て先となるノードを見つける
キーに対応するノードを見つける処理は次のように実装します：
// GetNode returns the node that should handle the given key
func (chr *ConsistentHashRing) GetNode(key string) *Node {
chr.mu.RLock()
defer chr.mu.RUnlock()
if len(chr.hashes) == 0 {
return nil
}
hash := hashFunction(key)
idx := chr.GetNextNodeIndex(hash)
return chr.nodes[chr.hashes[idx]]
}
// GetNextNodeIndex finds the next node in the ring after the given hash
func (chr *ConsistentHashRing) GetNextNodeIndex(hash uint32) int {
for i, h := range chr.hashes {
if h >= hash {
return i
}
}
return 0 // 最初のノードに戻る（リングの一周）
}
データの保存と取得
キーと値のペアをノードに保存し、取得する関数を実装します：
// StoreKey stores a key-value pair in the appropriate node
func (chr *ConsistentHashRing) StoreKey(key, val string) {
node := chr.GetNode(key)
if node != nil {
chr.mu.Lock()
node.Keys[key] = val
chr.mu.Unlock()
}
}
// RetrieveKey retrieves a value for a given key
func (chr *ConsistentHashRing) RetrieveKey(key string) (string, error) {
node := chr.GetNode(key)
if node == nil {
return "", errors.New("no node found")
}
chr.mu.RLock()
val, ok := node.Keys[key]
chr.mu.RUnlock()
if !ok {
return "", errors.New("key not found")
}
return val, nil
}
ノードの削除
ノードを削除する際は、そのノードが持っていたキーを次のノードに移す処理も行います：
// RemoveNode removes a node from the hash ring and redistributes its keys
func (chr *ConsistentHashRing) RemoveNode(id string) {
chr.mu.Lock()
defer chr.mu.Unlock()
hash := hashFunction(id)
node, exists := chr.nodes[hash]
if !exists {
return
}
// このノードが持っていたキーを次のノードに移す
idx := 0
for i, h := range chr.hashes {
if h == hash {
idx = i
break
}
}
nextIdx := (idx + 1) % len(chr.hashes)
nextNode := chr.nodes[chr.hashes[nextIdx]]
// すべてのキーを次のノードにコピー
for k, v := range node.Keys {
nextNode.Keys[k] = v
}
// ノードの削除
delete(chr.nodes, hash)
// ハッシュ配列からも削除
chr.hashes = append(chr.hashes[:idx], chr.hashes[idx+1:]...)
}
リング情報の表示
デバッグやモニタリングのために、ハッシュリングの情報を表示する関数も実装しておきます：
// PrintRing displays the current state of the hash ring
func (chr *ConsistentHashRing) PrintRing() {
chr.mu.RLock()
defer chr.mu.RUnlock()
for _, h := range chr.hashes {
fmt.Printf("Node: %s \t\t Hash: %d \t\t Total Keys: %d\n",
chr.nodes[h].ID, h, len(chr.nodes[h].Keys))
}
}
使用例
実際にこの実装を使用してみましょう：
func main() {
ring := NewConsistentHashRing()
// いくつかのノードを追加
servers := []string{"server1", "server2", "server3", "server4", "server5"}
for _, server := range servers {
ring.AddNode(server)
}
// いくつかのキーを保存
keys := []string{"user:1001", "user:1002", "product:5001", "order:8001", "order:8002"}
for i, key := range keys {
ring.StoreKey(key, fmt.Sprintf("value-%d", i+1))
}
// ノードとキーの分布状況を表示
fmt.Println("Initial distribution:")
ring.PrintRing()
// キーを取得
for _, key := range keys {
val, err := ring.RetrieveKey(key)
if err != nil {
fmt.Printf("Key %s not found: %v\n", key, err)
} else {
fmt.Printf("Retrieved %s = %s\n", key, val)
}
}
// 新しいノードを追加
fmt.Println("\nAdding a new node:")
ring.AddNode("server6")
ring.PrintRing()
// ノードを削除
fmt.Println("\nRemoving a node:")
ring.RemoveNode("server3")
ring.PrintRing()
}
このコードを実行すると、ノードの追加や削除を行っても、すべてのキーを再配置する必要がないことがわかります。これが通常のハッシュ関数と比較したConsistent Hashingの大きな利点です。
Consistent Hashingの実用的な応用例
Consistent Hashingは様々な実際のシステムで広く活用されています：
1. 分散キャッシュシステム
Memcachedや、Redis Clusterなどの分散キャッシュでは、キーをどのサーバーに保存するかを決めるために使われています。サーバーが追加・削除されてもキャッシュミスを最小限に抑えられるため、パフォーマンスの安定性が向上します。
2. ロードバランシング
API呼び出しやウェブリクエストを複数のサーバーに均等に分散するために使用されます。同じクライアントからの複数のリクエストを同じサーバーに割り当てる「スティッキーセッション」を実現するのに適しています。
3. データベースシャーディング
大規模データベースを複数のサーバーに分割（シャーディング）する際にも利用されます。レコードをどのデータベースサーバーに保存するかを決定するのに役立ちます。
4. コンテンツ配信ネットワーク（CDN）
Akamaiなどのサービスでは、コンテンツをどのエッジサーバーから配信するかを決める際に使用されています。
5. P2Pネットワーク
BitTorrentなどのP2Pネットワークでは、データの配置やピア検索にConsistent Hashingが使われています。
実装上の注意点と改善方法
基本的なConsistent Hashingには、ノードの追加・削除時のデータ移動を最小限に抑えるという利点がありますが、いくつかの課題もあります：
1. 仮想ノード（Virtual Nodes）の導入
基本的な実装では、ハッシュリング上のノード分布が偏る可能性があります。これを解決するために「仮想ノード（バーチャルノード）」という概念が導入されています。
各物理サーバーを複数の仮想ノードとしてリングに配置することで、より均一な分布を実現できます：
// 各サーバーに対して複数の仮想ノードを作成
func (chr *ConsistentHashRing) AddNodeWithReplicas(id string, replicas int) {
for i := 0; i < replicas; i++ {
virtualID := fmt.Sprintf("%s-%d", id, i)
hash := hashFunction(virtualID)
chr.nodes[hash] = &Node{
ID: id, // 元のサーバーIDを保持
Keys: make(map[string]string),
}
chr.hashes = append(chr.hashes, hash)
}
sort.Slice(chr.hashes, func(i, j int) bool { return chr.hashes[i] < chr.hashes[j] })
}
2. 負荷分散の最適化
サーバー間の負荷を均等に保つために、単純に「最も近いノード」ではなく、負荷状況も考慮した割り当てを行うことができます：
// 負荷状況も考慮したノード選択
func (chr *ConsistentHashRing) GetLeastLoadedNode(key string, maxLoad int) *Node {
chr.mu.RLock()
defer chr.mu.RUnlock()
if len(chr.hashes) == 0 {
return nil
}
hash := hashFunction(key)
// ハッシュリングを一周して、負荷が閾値以下のノードを探す
startIdx := chr.GetNextNodeIndex(hash)
idx := startIdx
for {
node := chr.nodes[chr.hashes[idx]]
if len(node.Keys) < maxLoad {
return node
}
idx = (idx + 1) % len(chr.hashes)
if idx == startIdx {
// 一周してきたら、最初に見つかったノードを返す
return chr.nodes[chr.hashes[startIdx]]
}
}
}
まとめ
Consistent Hashingは、分散システムにおけるデータ配置の問題を効率的に解決する重要なアルゴリズムです。
このアルゴリズムの主な利点は以下の通りです：
- サーバーの追加・削除時にデータ移動が最小限に抑えられる（通常は再配置が必要なのは全体のK/N程度、Kはキー数、Nはノード数）
- スケールアップ・ダウンが容易で、システムの可用性が向上する
- 仮想ノードの導入により、より均一な負荷分散が可能