ライブラリ

Distributed Cache（分散キャッシュ）

概要

分散キャッシュは、複数のマシンやノードにまたがってデータを保存するキャッシュシステムです。複数のコンピューターのRAMを統合し、単一のメモリ内データストアとしてプールし、データキャッシュとして高速なデータアクセスを提供します。

詳細

分散キャッシュシステムは、現代の大規模Webアプリケーションやマイクロサービスアーキテクチャにおいて不可欠な技術です。主要な分散キャッシュソリューションには、Redis（Remote Dictionary Server）、Memcached、Hazelcastなどがあります。Redisは高いパフォーマンスで知られるインメモリデータ構造ストアで、文字列、リスト、セット、ソート済みセット、ハッシュ、ビットマップ、地理空間インデックスなど豊富なデータ構造をサポートします。Memcachedはシンプルで強力な汎用分散メモリキャッシュシステムで、動的Webアプリケーションの高速化を目的としています。Hazelcastは分散インメモリデータグリッドで、キャッシュを超えて分散コンピューティング機能も提供します。分散キャッシュの主なメリットには、アプリケーション加速、スケーラビリティ、フォルトトレラント性、パフォーマンス向上があります。アーキテクチャパターンとしては、Embedded Cache、Client/Serverパターン、Sidecarパターンなどが利用されます。

メリット・デメリット

メリット

• スケーラビリティ: トラフィック増加に応じて追加のキャッシュサーバーを無停止で追加可能
• 高可用性: 1つのキャッシュサーバーが障害を起こしても、他のサーバーにリクエストを転送
• パフォーマンス向上: データがユーザーの近くに保存され、取得時間とレスポンス時間が改善
• フォルトトレラント性: データの複製により、単一障害点を排除
• 負荷分散: 複数のノードにデータとリクエストを分散し、システム全体の負荷を軽減
• 柔軟なアーキテクチャ: 様々なキャッシュパターン（cache-aside、read-through、write-through等）をサポート
• メモリ効率: 大量のデータを複数のマシンのRAMで効率的に管理

デメリット

• 複雑性: 単一ノードキャッシュと比較してシステムの複雑性が増加
• ネットワーク依存: ネットワークレイテンシがパフォーマンスに影響
• データ一貫性: 分散環境での一貫性管理が課題となる場合がある
• 設定・管理コスト: 複数ノードの設定、監視、メンテナンスが必要
• 障害の複雑化: ネットワーク分断やノード障害など、単一システムより障害パターンが複雑
• セキュリティ: 複数ノード間の通信セキュリティ確保が必要

主要リンク

• Redis分散キャッシング概要
• Hazelcast分散キャッシュ基礎
• マイクロサービス向けキャッシュパターン
• 分散キャッシュソリューション比較
• Redis vs Hazelcast比較
• Memcached公式サイト

書き方の例

Redis分散キャッシュの設定

import redis
from redis.sentinel import Sentinel

# Redis Sentinelを使用した高可用性設定
sentinel = Sentinel([
    ('localhost', 26379),
    ('localhost', 26380),
    ('localhost', 26381)
])

# マスターとスレーブの発見
master = sentinel.master_for('mymaster', socket_timeout=0.1)
slave = sentinel.slave_for('mymaster', socket_timeout=0.1)

# 書き込みはマスターに
master.set('key', 'value')

# 読み込みはスレーブから
value = slave.get('key')

# Redis Clusterの使用例
from rediscluster import RedisCluster

startup_nodes = [
    {"host": "127.0.0.1", "port": "7000"},
    {"host": "127.0.0.1", "port": "7001"},
    {"host": "127.0.0.1", "port": "7002"}
]

rc = RedisCluster(startup_nodes=startup_nodes, decode_responses=True)
rc.set("key", "value")
print(rc.get("key"))

Memcached分散キャッシュ（Python）

import pymemcache
from pymemcache.client.base import Client
from pymemcache.client.hash import HashClient

# 単一サーバー接続
client = Client(('localhost', 11211))
client.set('some_key', 'some_value')
result = client.get('some_key')

# 複数サーバーでの分散キャッシュ
servers = [
    ('127.0.0.1', 11211),
    ('127.0.0.1', 11212),
    ('127.0.0.1', 11213)
]

# コンシステントハッシュによる分散
hash_client = HashClient(servers)

# データの設定（自動的に適切なサーバーに分散）
hash_client.set('user:1001', {'name': 'Alice', 'age': 30})
hash_client.set('user:1002', {'name': 'Bob', 'age': 25})

# データの取得
user_data = hash_client.get('user:1001')
print(user_data)

# 複数データの一括操作
hash_client.set_many({
    'product:1': {'name': 'Laptop', 'price': 999},
    'product:2': {'name': 'Mouse', 'price': 29}
})

products = hash_client.get_many(['product:1', 'product:2'])

Hazelcast分散キャッシュ（Java）

import com.hazelcast.config.Config;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.map.IMap;

// Hazelcastクラスター設定
Config config = new Config();
config.setClusterName("dev-cluster");

// ネットワーク設定
config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
config.getNetworkConfig().getJoin().getTcpIpConfig()
    .setEnabled(true)
    .addMember("192.168.1.100")
    .addMember("192.168.1.101")
    .addMember("192.168.1.102");

// Hazelcastインスタンス作成
HazelcastInstance hazelcast = Hazelcast.newHazelcastInstance(config);

// 分散マップ（キャッシュ）の使用
IMap<String, String> cache = hazelcast.getMap("my-cache");

// データの保存（自動的にクラスター全体に分散）
cache.put("key1", "value1");
cache.put("key2", "value2");

// データの取得
String value = cache.get("key1");
System.out.println("Retrieved: " + value);

// TTL（Time To Live）付きでデータ保存
cache.put("temp-key", "temp-value", 30, TimeUnit.SECONDS);

// 分散実行例
hazelcast.getExecutorService("default").executeOnAllMembers(
    () -> System.out.println("実行ノード: " + 
        Hazelcast.getAllHazelcastInstances().iterator().next()
            .getCluster().getLocalMember())
);

Node.js でのRedis分散キャッシュ

const Redis = require('ioredis');

// Redis Clusterの設定
const cluster = new Redis.Cluster([
  {
    host: '127.0.0.1',
    port: 7000,
  },
  {
    host: '127.0.0.1',
    port: 7001,
  },
  {
    host: '127.0.0.1',
    port: 7002,
  }
], {
  redisOptions: {
    password: 'your-password'
  }
});

// キャッシュヘルパー関数
class DistributedCache {
  constructor(redisCluster) {
    this.redis = redisCluster;
  }

  async get(key) {
    try {
      const value = await this.redis.get(key);
      return value ? JSON.parse(value) : null;
    } catch (error) {
      console.error('Cache get error:', error);
      return null;
    }
  }

  async set(key, value, ttl = 3600) {
    try {
      await this.redis.setex(key, ttl, JSON.stringify(value));
      return true;
    } catch (error) {
      console.error('Cache set error:', error);
      return false;
    }
  }

  async del(key) {
    try {
      await this.redis.del(key);
      return true;
    } catch (error) {
      console.error('Cache delete error:', error);
      return false;
    }
  }

  async exists(key) {
    try {
      const result = await this.redis.exists(key);
      return result === 1;
    } catch (error) {
      console.error('Cache exists error:', error);
      return false;
    }
  }
}

// 使用例
const cache = new DistributedCache(cluster);

async function cacheExample() {
  // データキャッシュ
  await cache.set('user:1001', { 
    name: 'Alice', 
    email: '[email protected]',
    lastLogin: new Date()
  }, 1800); // 30分のTTL

  // データ取得
  const userData = await cache.get('user:1001');
  console.log('キャッシュされたユーザーデータ:', userData);

  // パターンマッチングによる削除
  const keys = await cluster.keys('user:*');
  if (keys.length > 0) {
    await cluster.del(...keys);
    console.log(`${keys.length}個のキーを削除しました`);
  }
}

cacheExample();

分散キャッシュアーキテクチャパターン

# Cache-Asideパターン（Lazy Loading）
class CacheAsidePattern:
    def __init__(self, cache_client, database):
        self.cache = cache_client
        self.db = database
    
    def get_user(self, user_id):
        # 1. キャッシュから確認
        cached_user = self.cache.get(f"user:{user_id}")
        if cached_user:
            return cached_user
        
        # 2. キャッシュミスの場合、データベースから取得
        user = self.db.get_user(user_id)
        if user:
            # 3. キャッシュに保存
            self.cache.set(f"user:{user_id}", user, ttl=3600)
        
        return user

# Write-Throughパターン
class WriteThroughPattern:
    def __init__(self, cache_client, database):
        self.cache = cache_client
        self.db = database
    
    def update_user(self, user_id, user_data):
        # 1. データベースに書き込み
        self.db.update_user(user_id, user_data)
        
        # 2. 同時にキャッシュも更新
        self.cache.set(f"user:{user_id}", user_data, ttl=3600)
        
        return user_data

# Write-Behindパターン（Write-Back）
import asyncio
from collections import deque

class WriteBehindPattern:
    def __init__(self, cache_client, database):
        self.cache = cache_client
        self.db = database
        self.write_queue = deque()
        self.start_background_writer()
    
    async def update_user(self, user_id, user_data):
        # 1. 即座にキャッシュを更新
        await self.cache.set(f"user:{user_id}", user_data, ttl=3600)
        
        # 2. 書き込みキューに追加（非同期でDBに書き込み）
        self.write_queue.append(('update_user', user_id, user_data))
        
        return user_data
    
    async def background_writer(self):
        while True:
            if self.write_queue:
                operation, user_id, user_data = self.write_queue.popleft()
                try:
                    await self.db.update_user(user_id, user_data)
                except Exception as e:
                    # エラー時はキューに戻す
                    self.write_queue.appendleft((operation, user_id, user_data))
                    print(f"DB書き込みエラー: {e}")
            
            await asyncio.sleep(1)  # 1秒間隔でチェック
    
    def start_background_writer(self):
        asyncio.create_task(self.background_writer())