分布式缓存

概念概述

分布式缓存（Distributed Caching）指的是：分布式缓存让缓存数据在众多应用实例间共享，形成一个可跨单个进程重启存活的逻辑缓存，但每次查找都增加网络往返。

从更严格的定义看，A distributed cache is a cache whose entries are stored on dedicated nodes or partitioned across them, so any application instance can read and write the same logical data set. 该概念在工程实践中的价值，在于它能帮助团队用统一的语言描述一个具体的质量属性或设计约束，从而在评审与协作中减少歧义。

为什么需要它

It is used when many horizontally scaled instances must serve the same hot data without each holding a divergent copy. 它与 Write-Behind Caching、Cache Partitioning via Consistent Hashing、Edge Caching at the CDN 等概念经常一起出现，用于解决相近的工程问题。理解它的适用场景，有助于在多种备选方案之间做出有依据的取舍，而不是凭直觉选择技术方案。

常见误解

• a distributed cache is automatically consistent across nodes
• an in-process cache and a distributed cache are interchangeable

需要避免的反模式

• treating a distributed cache as durable storage
• using it without a fallback when it becomes unavailable

实践要点

1. Because entries live on remote nodes, every lookup adds a network round trip, which is acceptable for large shared objects but wasteful for tiny hot values.
2. Partitioning and replication decide what happens when a node fails, and an eviction on one node affects all clients that read that key, unlike a local cache.
3. Treat the cache as best-effort: a miss or an outage must fall back to the source of truth rather than fail the request.

参考来源

https://redis.io/docs/latest/operate/oss_and_stack/management/scaling/

代码示例

interface CacheEntry<T> {
  value: T;
  expiresAt: number;
}

function Cached<T>(ttlMs: number) {
  return function (
    target: object,
    propertyKey: string,
    descriptor: TypedPropertyDescriptor<(...args: never[]) => Promise<T>>,
  ) {
    const original = descriptor.value!;
    const store = new Map<string, CacheEntry<T>>();

    descriptor.value = async function (this: unknown, ...args: never[]) {
      const key = propertyKey + ':' + JSON.stringify(args);
      const hit = store.get(key);
      const now = Date.now();

      if (hit && hit.expiresAt > now) {
        (this as { __cacheHits?: number }).__cacheHits ??= 0;
        return hit.value;
      }

      const value = await original.apply(this, args);
      store.set(key, { value, expiresAt: now + ttlMs });
      return value;
    };

    return descriptor;
  };
}

class ExchangeRates {
  private hits = 0;

  @Cached(30_000)
  async ratesFor(base: string): Promise<Record<string, number>> {
    const res = await fetch('https://api.example.com/rates?base=' + base);
    return res.json();
  }

  @Cached(300_000)
  async currencies(): Promise<string[]> {
    return ['USD', 'EUR', 'JPY'];
  }
}

const rates = new ExchangeRates();
const first = await rates.ratesFor('USD');
const second = await rates.ratesFor('USD');
console.log('cached identity:', first === second);

import asyncio
import functools
import json

import redis.asyncio as redis

pool = redis.ConnectionPool.from_url('redis://localhost:6379/0')
client = redis.Redis(connection_pool=pool)


def async_cache(ttl: int = 60):
    def decorator(fn):
        @functools.wraps(fn)
        async def wrapper(*args, **kwargs):
            key = f'cache:{fn.__module__}:{fn.__qualname__}:{args!r}:{sorted(kwargs.items())!r}'
            cached = await client.get(key)
            if cached is not None:
                return json.loads(cached)

            lock_key = key + ':lock'
            # 简化版单飞：拿不到锁的并发请求稍候后直接读缓存
            got_lock = await client.set(lock_key, '1', nx=True, ex=ttl)
            if got_lock:
                try:
                    value = await fn(*args, **kwargs)
                    await client.set(key, json.dumps(value), ex=ttl)
                    return value
                finally:
                    await client.delete(lock_key)

            await asyncio.sleep(0.05)
            cached = await client.get(key)
            if cached is not None:
                return json.loads(cached)
            return await fn(*args, **kwargs)

        return wrapper

    return decorator


@async_cache(ttl=90)
async def exchange_rates(base: str) -> dict:
    print(f'cache miss: fetching rates for {base}')
    await asyncio.sleep(0.3)
    return {'base': base, 'usd': 1.0, 'eur': 0.92}


async def main() -> None:
    first, second = await asyncio.gather(
        exchange_rates('usd'), exchange_rates('usd')
    )
    print('single flight result equal:', first == second)
    await client.aclose()


asyncio.run(main())