キャッシュライブラリ

Ristretto

概要

RistrettoはGo言語向けの高性能なメモリ内キャッシュライブラリで、競合状態を回避し、優れたスループット性能とヒット率を提供する、並行処理に最適化されたキャッシュソリューションです。

詳細

Ristretto（リストレット）は、パフォーマンスと正確性に焦点を当てて構築されたGo言語向けの高速で並行処理対応のキャッシュライブラリです。Dgraphでの競合のないキャッシュの必要性から生まれました。高いヒット率を提供する独自の許可/排除ポリシーのペアリング、競合を管理するための様々な技術による高速スループット、多数のgoroutineを使用してもスループットの劣化がほとんどない完全な並行処理を特徴としています。排除ポリシーにはSampledLFU（正確なLRUと同等でありながら検索・データベーストレースでより良いパフォーマンス）、許可ポリシーにはTinyLFU（カウンターあたり12ビットという少ないメモリオーバーヘッドで追加のパフォーマンス）を採用しています。コストベースの排除機能により、価値があると判断された大きな新しいアイテムは複数の小さなアイテムを排除できます。オプションのパフォーマンスメトリクス（スループット、ヒット率、その他の統計）も提供しますが、10%のスループットパフォーマンスオーバーヘッドがあるため設定フラグとなっています。バッチ処理と結果整合性の組み合わせによってスループットパフォーマンスが、優れた許可ポリシーとSampledLFU排除ポリシーによってヒット率パフォーマンスが実現されています。競合抵抗に最適化されており、重い並行負荷の下でも非常に良いパフォーマンスを発揮し、一貫して高いヒット率を提供します。

メリット・デメリット

メリット

• 高性能: 優れたスループットと低レイテンシ
• 高ヒット率: TinyLFUとSampledLFU による最適化されたキャッシュ効率
• 完全並行: goroutine数に関係なく高いパフォーマンスを維持
• 競合抵抗: 重い並行負荷下でも一貫したパフォーマンス
• 柔軟な設定: コストベース排除やメトリクス機能
• メモリ効率: 最小限のメモリオーバーヘッド
• 型安全: Go generics サポート（v2.x）

デメリット

• Go専用: Go言語でのみ利用可能
• メトリクスオーバーヘッド: メトリクス有効時は10%のパフォーマンス低下
• 設定複雑性: 最適化のための設定パラメータが多い
• メモリ境界: メモリベースのキャッシュのため永続化なし
• 学習コスト: 高度な機能の習得に時間が必要

主要リンク

• Ristretto GitHub リポジトリ
• Ristretto Go Package ドキュメント
• Ristretto 紹介ブログ
• Dgraph 公式サイト
• Go 公式サイト

書き方の例

インストールと基本設定

# Go modules環境でのインストール
go get github.com/dgraph-io/ristretto

package main

import (
    "fmt"
    "time"
    
    "github.com/dgraph-io/ristretto"
)

func main() {
    // 基本的なキャッシュの作成
    cache, err := ristretto.NewCache(&ristretto.Config{
        NumCounters: 1e7,     // 追跡するキー数 (10M)
        MaxCost:     1 << 30, // キャッシュの最大コスト (1GB)
        BufferItems: 64,      // Getバッファあたりのキー数
    })
    if err != nil {
        panic(err)
    }
    defer cache.Close()
    
    // 基本的な操作
    cache.Set("key", "value", 1)
    
    // 値の取得
    value, found := cache.Get("key")
    if found {
        fmt.Println("キャッシュヒット:", value)
    }
    
    // キャッシュのクリア
    cache.Clear()
}

型安全なジェネリクス使用（v2.x）

package main

import (
    "fmt"
    "github.com/dgraph-io/ristretto/v2"
)

type User struct {
    ID   int
    Name string
    Email string
}

func main() {
    // 型安全なキャッシュの作成
    cache, err := ristretto.NewCache[string, *User](&ristretto.Config[string, *User]{
        NumCounters: 1000,
        MaxCost:     100,
        BufferItems: 64,
    })
    if err != nil {
        panic(err)
    }
    defer cache.Close()
    
    // ユーザーデータのキャッシュ
    user := &User{
        ID:   1,
        Name: "Alice",
        Email: "[email protected]",
    }
    
    // コストを考慮した設定
    cache.Set("user:1", user, 1)
    
    // 型安全な取得
    if cachedUser, found := cache.Get("user:1"); found {
        fmt.Printf("ユーザー: %+v\n", cachedUser)
    }
    
    // 削除
    cache.Del("user:1")
}

コストベースの排除

package main

import (
    "fmt"
    "github.com/dgraph-io/ristretto"
)

func main() {
    cache, err := ristretto.NewCache(&ristretto.Config{
        NumCounters: 1000,
        MaxCost:     100, // 総コスト100まで
        BufferItems: 64,
    })
    if err != nil {
        panic(err)
    }
    defer cache.Close()
    
    // 異なるコストでアイテムを保存
    cache.Set("small-item", "data1", 1)   // コスト1
    cache.Set("medium-item", "data2", 5)  // コスト5
    cache.Set("large-item", "data3", 20)  // コスト20
    
    // 大きなアイテムが小さなアイテムを排除する可能性
    cache.Set("huge-item", "data4", 80)   // コスト80
    
    // 各アイテムの存在確認
    items := []string{"small-item", "medium-item", "large-item", "huge-item"}
    
    for _, key := range items {
        if _, found := cache.Get(key); found {
            fmt.Printf("%s: キャッシュに存在\n", key)
        } else {
            fmt.Printf("%s: 排除された\n", key)
        }
    }
}

TTL（有効期限）の設定

package main

import (
    "fmt"
    "time"
    
    "github.com/dgraph-io/ristretto"
)

func main() {
    cache, err := ristretto.NewCache(&ristretto.Config{
        NumCounters: 1000,
        MaxCost:     100,
        BufferItems: 64,
    })
    if err != nil {
        panic(err)
    }
    defer cache.Close()
    
    // TTL付きでアイテムを設定
    cache.SetWithTTL("session:123", "session-data", 1, 5*time.Second)
    cache.SetWithTTL("temp-data", "temporary", 1, 2*time.Second)
    
    // 即座にチェック
    if value, found := cache.Get("session:123"); found {
        fmt.Printf("セッション: %v\n", value)
    }
    
    // 3秒待機
    time.Sleep(3 * time.Second)
    
    if _, found := cache.Get("temp-data"); !found {
        fmt.Println("temp-data は期限切れで削除されました")
    }
    
    if value, found := cache.Get("session:123"); found {
        fmt.Printf("セッション（3秒後）: %v\n", value)
    }
    
    // さらに3秒待機（合計6秒）
    time.Sleep(3 * time.Second)
    
    if _, found := cache.Get("session:123"); !found {
        fmt.Println("session:123 は期限切れで削除されました")
    }
}

メトリクス機能

package main

import (
    "fmt"
    "github.com/dgraph-io/ristretto"
)

func main() {
    // メトリクス有効なキャッシュ
    cache, err := ristretto.NewCache(&ristretto.Config{
        NumCounters: 1000,
        MaxCost:     100,
        BufferItems: 64,
        Metrics:     true, // メトリクス有効化
    })
    if err != nil {
        panic(err)
    }
    defer cache.Close()
    
    // テストデータの挿入と取得
    for i := 0; i < 100; i++ {
        key := fmt.Sprintf("key:%d", i)
        cache.Set(key, fmt.Sprintf("value:%d", i), 1)
    }
    
    // キャッシュヒット/ミスのテスト
    hitCount := 0
    missCount := 0
    
    // 存在するキーへのアクセス
    for i := 0; i < 50; i++ {
        key := fmt.Sprintf("key:%d", i)
        if _, found := cache.Get(key); found {
            hitCount++
        } else {
            missCount++
        }
    }
    
    // 存在しないキーへのアクセス
    for i := 100; i < 150; i++ {
        key := fmt.Sprintf("key:%d", i)
        if _, found := cache.Get(key); found {
            hitCount++
        } else {
            missCount++
        }
    }
    
    // メトリクス取得
    metrics := cache.Metrics
    fmt.Printf("ヒット数: %d\n", hitCount)
    fmt.Printf("ミス数: %d\n", missCount)
    fmt.Printf("キャッシュメトリクス: %+v\n", metrics)
}

並行処理での使用

package main

import (
    "fmt"
    "sync"
    "time"
    
    "github.com/dgraph-io/ristretto"
)

func main() {
    cache, err := ristretto.NewCache(&ristretto.Config{
        NumCounters: 10000,
        MaxCost:     1000,
        BufferItems: 64,
        Metrics:     true,
    })
    if err != nil {
        panic(err)
    }
    defer cache.Close()
    
    var wg sync.WaitGroup
    numGoroutines := 100
    operationsPerGoroutine := 1000
    
    // 複数のgoroutineで並行書き込み
    wg.Add(numGoroutines)
    for i := 0; i < numGoroutines; i++ {
        go func(goroutineID int) {
            defer wg.Done()
            
            for j := 0; j < operationsPerGoroutine; j++ {
                key := fmt.Sprintf("goroutine:%d:key:%d", goroutineID, j)
                value := fmt.Sprintf("value:%d:%d", goroutineID, j)
                cache.Set(key, value, 1)
            }
        }(i)
    }
    
    // 並行読み取り
    wg.Add(numGoroutines)
    for i := 0; i < numGoroutines; i++ {
        go func(goroutineID int) {
            defer wg.Done()
            
            for j := 0; j < operationsPerGoroutine; j++ {
                key := fmt.Sprintf("goroutine:%d:key:%d", goroutineID, j)
                if value, found := cache.Get(key); found {
                    _ = value // 値を使用
                }
            }
        }(i)
    }
    
    wg.Wait()
    
    fmt.Printf("並行処理完了 - %d goroutines, %d operations each\n", 
               numGoroutines, operationsPerGoroutine)
    fmt.Printf("メトリクス: %+v\n", cache.Metrics)
}

Webアプリケーションでの使用例

package main

import (
    "encoding/json"
    "fmt"
    "net/http"
    "strconv"
    "time"
    
    "github.com/dgraph-io/ristretto"
)

type UserService struct {
    cache *ristretto.Cache
}

type User struct {
    ID    int    `json:"id"`
    Name  string `json:"name"`
    Email string `json:"email"`
}

func NewUserService() *UserService {
    cache, err := ristretto.NewCache(&ristretto.Config{
        NumCounters: 1e7,
        MaxCost:     1 << 30,
        BufferItems: 64,
        Metrics:     true,
    })
    if err != nil {
        panic(err)
    }
    
    return &UserService{cache: cache}
}

func (us *UserService) GetUser(id int) (*User, error) {
    cacheKey := fmt.Sprintf("user:%d", id)
    
    // キャッシュから取得を試行
    if cachedUser, found := us.cache.Get(cacheKey); found {
        if user, ok := cachedUser.(*User); ok {
            return user, nil
        }
    }
    
    // データベースから取得（シミュレート）
    user := &User{
        ID:    id,
        Name:  fmt.Sprintf("User %d", id),
        Email: fmt.Sprintf("user%[email protected]", id),
    }
    
    // キャッシュに保存（1時間TTL）
    us.cache.SetWithTTL(cacheKey, user, 1, time.Hour)
    
    return user, nil
}

func (us *UserService) InvalidateUser(id int) {
    cacheKey := fmt.Sprintf("user:%d", id)
    us.cache.Del(cacheKey)
}

func (us *UserService) GetCacheMetrics() interface{} {
    return us.cache.Metrics
}

func main() {
    userService := NewUserService()
    
    http.HandleFunc("/user/", func(w http.ResponseWriter, r *http.Request) {
        idStr := r.URL.Path[len("/user/"):]
        id, err := strconv.Atoi(idStr)
        if err != nil {
            http.Error(w, "Invalid user ID", http.StatusBadRequest)
            return
        }
        
        user, err := userService.GetUser(id)
        if err != nil {
            http.Error(w, "User not found", http.StatusNotFound)
            return
        }
        
        w.Header().Set("Content-Type", "application/json")
        json.NewEncoder(w).Encode(user)
    })
    
    http.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
        metrics := userService.GetCacheMetrics()
        w.Header().Set("Content-Type", "application/json")
        json.NewEncoder(w).Encode(metrics)
    })
    
    fmt.Println("サーバーがポート8080で起動しました")
    http.ListenAndServe(":8080", nil)
}

データベースクエリキャッシュ

package main

import (
    "fmt"
    "time"
    
    "github.com/dgraph-io/ristretto"
)

type QueryCache struct {
    cache *ristretto.Cache
}

func NewQueryCache() *QueryCache {
    cache, err := ristretto.NewCache(&ristretto.Config{
        NumCounters: 1e7,
        MaxCost:     1 << 30,
        BufferItems: 64,
    })
    if err != nil {
        panic(err)
    }
    
    return &QueryCache{cache: cache}
}

func (qc *QueryCache) Execute(query string, ttl time.Duration) (interface{}, error) {
    // クエリハッシュをキーとして使用
    cacheKey := fmt.Sprintf("query:%s", query)
    
    // キャッシュから取得
    if result, found := qc.cache.Get(cacheKey); found {
        fmt.Println("キャッシュヒット:", query)
        return result, nil
    }
    
    // データベースクエリをシミュレート
    fmt.Println("データベースクエリ実行:", query)
    time.Sleep(100 * time.Millisecond) // DB遅延をシミュレート
    
    // 結果をシミュレート
    result := map[string]interface{}{
        "query":     query,
        "timestamp": time.Now(),
        "data":      []string{"row1", "row2", "row3"},
    }
    
    // キャッシュに保存
    qc.cache.SetWithTTL(cacheKey, result, 1, ttl)
    
    return result, nil
}

func main() {
    queryCache := NewQueryCache()
    
    queries := []string{
        "SELECT * FROM users WHERE active = true",
        "SELECT COUNT(*) FROM orders WHERE date > '2023-01-01'",
        "SELECT * FROM products WHERE category = 'electronics'",
    }
    
    // 各クエリを複数回実行
    for i := 0; i < 3; i++ {
        fmt.Printf("\n=== ラウンド %d ===\n", i+1)
        
        for _, query := range queries {
            result, err := queryCache.Execute(query, 5*time.Minute)
            if err != nil {
                fmt.Printf("エラー: %v\n", err)
                continue
            }
            
            fmt.Printf("結果: %v\n", result)
        }
        
        // 少し待機
        time.Sleep(1 * time.Second)
    }
}

カスタム排除ポリシー

package main

import (
    "fmt"
    "github.com/dgraph-io/ristretto"
)

// カスタムコスト計算関数
func calculateCost(value interface{}) int64 {
    switch v := value.(type) {
    case string:
        return int64(len(v))
    case []byte:
        return int64(len(v))
    case map[string]interface{}:
        return 10 // マップは固定コスト
    default:
        return 1 // デフォルト
    }
}

func main() {
    cache, err := ristretto.NewCache(&ristretto.Config{
        NumCounters: 1000,
        MaxCost:     1000, // 1000バイト制限
        BufferItems: 64,
        Metrics:     true,
    })
    if err != nil {
        panic(err)
    }
    defer cache.Close()
    
    // 異なるサイズのデータを保存
    smallData := "small"
    mediumData := "this is a medium sized string"
    largeData := "this is a very large string that takes up much more space than the others"
    
    // 実際のデータサイズに基づいてコストを設定
    cache.Set("small", smallData, calculateCost(smallData))
    cache.Set("medium", mediumData, calculateCost(mediumData))
    cache.Set("large", largeData, calculateCost(largeData))
    
    // マップデータ
    mapData := map[string]interface{}{
        "key1": "value1",
        "key2": "value2",
        "key3": "value3",
    }
    cache.Set("map", mapData, calculateCost(mapData))
    
    // キャッシュの状態確認
    items := []string{"small", "medium", "large", "map"}
    for _, key := range items {
        if value, found := cache.Get(key); found {
            cost := calculateCost(value)
            fmt.Printf("%s (コスト: %d): キャッシュに存在\n", key, cost)
        } else {
            fmt.Printf("%s: 排除された\n", key)
        }
    }
    
    fmt.Printf("\nキャッシュメトリクス: %+v\n", cache.Metrics)
}