scala

Scala 缓存与 Redis 集成：Caffeine、Redis4cats 与缓存模式

系统讲解 Scala 缓存体系：本地 Caffeine 与分布式 Redis 的分层设计、Redis4cats 类型安全客户端、Cache-Aside 等四种缓存模式、穿透击穿雪崩的治理手段、分布式锁与限流、序列化编解码、效果系统集成，并附生产监控与容量规划实践。

Leeting Yan 9 分钟阅读 4019 字

缓存是性能优化的第一杠杆，也是线上事故的第一来源。Scala 生态里本地缓存首选 Caffeine，分布式缓存通过 Redis4cats 接入 Redis，两者组合成 L1/L2 分层；但真正难的不是调用 API，而是想清楚「什么时候写、什么时候失效、失效时谁来兜底」。本文从分层策略讲起，逐层落到本地缓存、Redis 客户端、缓存模式与一致性治理，最后给出效果系统集成与生产监控的完整实践。

前置：/scala-database-access/（数据库访问基础）、/scala-performance-jvm/（JVM 性能调优）。

目录

1. 缓存分层与策略

缓存的第一原则是「越靠近计算越快，容量越小」。L1 本地缓存（堆内）延迟纳秒级、容量受 JVM 堆限制、多实例之间天然不一致；L2 分布式缓存（Redis）延迟亚毫秒、容量可横向扩展、全局一致但多一次网络往返。合理架构是 L1 挡热点、L2 兜全量、DB 作为唯一真相源。

分层职责
- L1 本地 Caffeine：单机热点、极小 TTL、可容忍短暂不一致
- L2 Redis：跨实例共享、中等 TTL、DB 前的最后一道防线
- DB：唯一真相源，缓存只是派生数据

关键指标
- 命中率 hitRate = hits / (hits + misses)，目标 90% 以上
- 回源 QPS = 总 QPS 乘以 (1 减 命中率)
- 平均延迟 = hitRate 乘 cacheLatency 加 (1 减 hitRate) 乘 dbLatency
淘汰策略含义适用场景
LRU淘汰最久未访问的条目通用访问模式
LFU淘汰访问频次最低的条目热点长尾分明
W-TinyLFUCaffeine 默认，兼顾频率与新鲜度追求高命中率
TTL到期即失效强时效数据
Random随机淘汰实现简单的近似
def withJitter(base: FiniteDuration, ratio: Double = 0.2): FiniteDuration = {
  val jitterMs = (base.toMillis * ratio).toLong
  base + scala.util.Random.nextLong(jitterMs + 1).millis
}

工程要点：TTL 是「一致性预算」而不是随便填的数字，它决定了数据最坏情况下能陈旧多久。命中率低于 80% 时先怀疑 key 设计与 TTL 过短，而不是急着加大 Redis 内存——低命中率的高内存占用只是把浪费放大了。

2. Caffeine 本地缓存

Caffeine 是 JVM 上命中率最高的本地缓存，基于 W-TinyLFU 淘汰算法；Scala 侧可以用 scala-cache（ScalaCache）包出更函数式的 API，也可以直接调用 Caffeine 的 Java API（Scala 调用无痛）。

核心能力
- maximumSize / maximumWeight 加 weigher：容量控制
- expireAfterWrite / expireAfterAccess：时间维度失效
- refreshAfterWrite：异步刷新，读路径不阻塞
- LoadingCache：未命中自动加载，天然防击穿
- recordStats：命中率、加载耗时等统计

import com.github.benmanes.caffeine.cache.{Caffeine, LoadingCache}
import java.util.concurrent.TimeUnit

final case class User(id: Long, name: String)

val cache: LoadingCache[Long, User] =
  Caffeine.newBuilder()
    .maximumSize(10_000)
    .expireAfterWrite(5, TimeUnit.MINUTES)
    .refreshAfterWrite(1, TimeUnit.MINUTES)
    .recordStats()
    .build((id: Long) => UserRepo.load(id))

val u     = cache.get(1L)
val stats = cache.stats()
println(s"hitRate=${stats.hitRate()} avgLoad=${stats.averageLoadPenalty()}")

工程要点：refreshAfterWrite 只在有读请求时触发异步刷新，让读永远拿到旧值而不阻塞，本质上就是「逻辑过期」的内置实现；但刷新任务默认跑在 ForkJoinPool.commonPool，生产环境应显式传入专用 Executor，避免与业务任务争抢线程。

3. Redis 集成 Redis4cats

Redis4cats 是 Cats Effect 生态的 Redis 客户端（底层 Lettuce），提供类型安全命令、Resource 化连接与 fs2 流式操作，是 Scala 3 时代接入 Redis 的主流选择。

连接与命令
- Redis[IO].utf8(uri)：Resource 管理连接生命周期
- RedisCommands：get/set/setEx/del 等类型安全方法
- codec：RedisCodec[K, V] 决定 key 与 value 的编解码
- pipeline：批量往返，降低 RTT 开销
- cluster / sentinel：RedisCluster.fromNodes 支持集群

import dev.profunktor.redis4cats.Redis
import dev.profunktor.redis4cats.effect.Log.Stdout.given

val redis: Resource[IO, RedisCommands[IO, String, String]] =
  Redis[IO].utf8("redis://localhost:6379")

val run: IO[Unit] = redis.use { r =>
  for {
    _ <- r.set("user:1:name", "leeting")
    _ <- r.expire("user:1:name", 5.minutes)
    v <- r.get("user:1:name")
    _ <- IO.println(v) // Some(leeting)
  } yield ()
}

// pipeline：多条命令合并为一次网络往返
val batch: IO[Unit] = redis.use { r =>
  r.pipeline(r.set("k1", "a"), r.set("k2", "b"), r.set("k3", "c")).void
}

工程要点：连接必须用 Resource 管理，Redis[IO].utf8(uri) 在 use 结束后会正确关闭 Lettuce 连接与线程池；把客户端做成全局单例再手动 unsafeRunSync 取出来，是生产环境连接泄漏最常见的来源。

4. 缓存模式

四种经典模式的差异集中在两点：谁负责读写缓存、写操作发生在什么时候。

模式读路径写路径一致性
Cache-Aside应用查缓存，未命中回源并回填先写 DB 再删缓存最终一致
Read-Through缓存组件自己负责回源同 Cache-Aside最终一致
Write-Through同 Read-Through写缓存时同步写 DB较强
Write-Behind同 Read-Through只写缓存，异步刷 DB弱，可能丢数据
def key(id: Long) = s"v1:user:$id"

// Cache-Aside 读：先查缓存，未命中回源并回填
def getUser(id: Long): IO[Option[User]] =
  redis.get(key(id)).flatMap {
    case Some(json) => IO.pure(decode[User](json).toOption)
    case None =>
      UserRepo.find(id).flatMap {
        case Some(u) => redis.setEx(key(id), encode(u), 5.minutes).as(u.some)
        case None    => IO.pure(none[User])
      }
  }

// Cache-Aside 写：先更新 DB，再删除缓存
def updateUser(u: User): IO[Unit] =
  UserRepo.update(u) >> redis.del(key(u.id)).void

延迟双删用于缓解「读请求把旧值回填」的竞态：先删缓存、写 DB、延迟几百毫秒后再删一次。

def doubleDelete(u: User): IO[Unit] =
  redis.del(key(u.id)) >> UserRepo.update(u) >>
    IO.sleep(500.millis) >> redis.del(key(u.id)).void

工程要点：Cache-Aside 的推荐顺序是「先写 DB 后删缓存」而不是「先删缓存后写 DB」，后者在高并发读下极易被旧值回填。删除永远比更新安全：更新缓存会引入并发写互相覆盖，删除只会触发一次回源。

5. 一致性难题

三大经典故障
- 穿透：查询根本不存在的数据，每次都穿透到 DB
- 击穿：单个热 key 过期瞬间，大量请求同时回源
- 雪崩：大批 key 同时过期，或缓存整体不可用

应对手段
- 穿透：空值缓存（短 TTL）加 布隆过滤器前置拦截
- 击穿：互斥锁回源 / 逻辑过期 / refreshAfterWrite
- 雪崩：TTL 加随机抖动 / 多级缓存 / 熔断降级

// 布隆过滤器：在缓存之前拦掉必然不存在的 key
val bloom: BloomFilter[CharSequence] =
  BloomFilter.create[CharSequence](Funnels.stringFunnel(StandardCharsets.UTF_8), 1_000_000, 0.01)

def getWithBloom(id: Long): IO[Option[User]] =
  if (!bloom.mightContain(id.toString)) IO.pure(none[User]) else getUser(id)

互斥锁回源（防击穿）：只让一个请求去重建缓存，其余短暂自旋等待。

def getWithMutex(id: Long): IO[Option[User]] =
  redis.get(key(id)).flatMap {
    case s @ Some(_) => IO.pure(s.flatMap(decode[User](_).toOption))
    case None =>
      redis.setNx(s"lock:user:$id", "1", 3.seconds).flatMap {
        case true  => getUser(id)                             // 抢到锁，回源重建
        case false => IO.sleep(50.millis) >> getWithMutex(id) // 自旋等待
      }
  }

工程要点：逻辑过期（缓存永不物理过期，值里带一个逻辑过期时间，过期后由后台异步重建）能彻底消除击穿，代价是读到的可能是陈旧数据——它把一致性换成了可用性，适合商品详情这类能容忍秒级陈旧的场景，不适合账户余额。

6. 分布式锁与限流

Redlock 的争议在于它依赖各节点时钟推进的假设：Martin Kleppmann 认为它无法提供强一致保证，Redis 作者 antirez 认为在合理假设下足够用。工程结论是——非金融场景用「单实例 SET NX PX + 唯一 token + Lua 释放」即可，强一致场景请改用 etcd 或 ZooKeeper。

val UnlockScript: String =
  """if redis.call('get', KEYS[1]) == ARGV[1] then
    |  return redis.call('del', KEYS[1])
    |else
    |  return 0
    |end""".stripMargin

def withLock[A](lockKey: String, ttl: FiniteDuration)(fa: IO[A]): IO[A] =
  for {
    token <- IO(java.util.UUID.randomUUID().toString)
    ok    <- redis.setNx(lockKey, token, ttl)
    res   <- if (ok) fa.guarantee(redis.eval(UnlockScript, List(lockKey), List(token)).void)
             else IO.raiseError(new RuntimeException("lock busy"))
  } yield res

令牌桶限流同样用 Lua 保证「读令牌、补令牌、扣令牌」三步的原子性：

val TokenBucket: String =
  """local tokens = tonumber(redis.call('hget', KEYS[1], 'tokens') or ARGV[1])
    |local ts = tonumber(redis.call('hget', KEYS[1], 'ts') or ARGV[3])
    |tokens = math.min(tonumber(ARGV[1]), tokens + (tonumber(ARGV[3]) - ts) * tonumber(ARGV[2]))
    |if tokens >= 1 then
    |  redis.call('hmset', KEYS[1], 'tokens', tokens - 1, 'ts', ARGV[3])
    |  return 1
    |end
    |return 0""".stripMargin

工程要点：释放锁必须「先校验 token 再删」，否则会误删他人持有的锁；锁的 TTL 要大于业务最长执行时间，并在业务执行期间续期（watchdog），否则业务还没跑完锁就过期，临界区保护形同虚设。

7. 序列化与编解码

选型维度
- 性能：jsoniter 与 circe 量级相近，Protobuf 编解码更快
- 可读性：JSON 人类可读、Protobuf 二进制不可读
- 兼容性：Protobuf 靠字段编号演进、JSON 靠可选字段
- 体积：Protobuf 约为 JSON 的三分之一到五分之一
- 压缩：大 value 可用 LZ4/Snappy，本质是 CPU 换带宽

import io.circe.syntax.*
import io.circe.parser.decode
import io.circe.generic.auto.*

def encode[A: io.circe.Encoder](a: A): String = a.asJson.noSpaces
def decodeAs[A: io.circe.Decoder](s: String): Either[io.circe.Error, A] = decode[A](s)

key 命名与版本化是很多团队忽略的一环：

object Keys {
  private val v = "v1"
  def user(id: Long): String  = s"$v:user:$id"
  def order(id: Long): String = s"$v:order:$id"
  def lock(k: String): String = s"lock:$k"
}

工程要点：key 前缀加版本号（如 v1:user:1）能把「模型变更」变成一次平滑迁移——新代码写 v2、灰度期双读、旧 key 自然过期；没有版本号时改一次 case class 就可能造成反序列化全量失败，且无法回滚。

8. 与效果系统集成

集成方式
- Cats Effect Resource：连接与缓存的获取、释放
- ZIO ZLayer：把缓存客户端做成可注入的服务
- fs2：流式读取大批量缓存数据
- Tagless Final 接口：便于测试替换与统一切面

trait Cache[F[_]] {
  def get(k: String): F[Option[String]]
  def set(k: String, v: String, ttl: FiniteDuration): F[Unit]
  def del(k: String): F[Unit]
}

object Cache {
  def redis[F[_]: Async](uri: String): Resource[F, Cache[F]] =
    Redis[F].utf8(uri).map { r =>
      new Cache[F] {
        def get(k: String): F[Option[String]] = r.get(k)
        def set(k: String, v: String, ttl: FiniteDuration): F[Unit] = r.setEx(k, v, ttl)
        def del(k: String): F[Unit] = r.del(k).void
      }
    }
}

ZIO 侧把客户端包成 ZLayer 通过环境注入，fs2 侧则用于预热与批量导出：

val cacheLayer: ZLayer[Any, Throwable, Cache[Task]] =
  ZLayer.scoped(ZIO.acquireRelease(openRedis)(closeRedis).map(new CacheImpl(_)))

val entries: fs2.Stream[IO, (String, Option[String])] =
  fs2.Stream.emits(keys).covary[IO].evalMap(k => redis.get(k).map(k -> _))

工程要点：把缓存抽象成 Tagless Final 的 Cache[F] 之后，测试里可以换成纯内存 Map 实现，无需真实 Redis；同时它天然成为加监控、加降级、加熔断的统一切面。

9. 生产实践

监控指标
- 业务侧：命中率、回源 QPS、平均延迟、错误率
- Redis 侧：keyspace_hits / keyspace_misses、内存使用率、慢查询
- 本地缓存：Caffeine stats、GC 暂停时间
容量规划
- 单 key 平均大小 乘 期望 key 数 乘 1.5 冗余系数
- 设置 maxmemory 与淘汰策略（通常 allkeys-lru）
热 key 治理
- 本地缓存挡一层、key 拆分、读写分离
持久化
- RDB：定时快照，恢复快、可能丢分钟级数据
- AOF：appendfsync everysec，丢失窗口约 1 秒

def record(reg: MeterRegistry, hit: Boolean): Unit =
  Counter.builder("cache.requests")
    .tag("result", if (hit) "hit" else "miss")
    .register(reg)
    .increment()

测试时用 Testcontainers 起真实 Redis，避免 mock 掩盖 Lua 脚本与序列化问题：

val container = new GenericContainer("redis:7-alpine").withExposedPorts(6379)
container.start()
val uri = s"redis://${container.getHost}:${container.getMappedPort(6379)}"

工程要点：热 key 的典型信号是「单个 Redis 实例 CPU 明显高于其他」或「某个 key 的 QPS 占比异常」，本地缓存加短 TTL 是性价比最高的解法；持久化则按「能丢多少数据」来选，RDB 与 AOF 可以同时开启，用 AOF 保安全、用 RDB 加快重启。

10. 速查表与一句话记忆

问题结论
本地缓存选型Caffeine
分布式缓存Redis 加 Redis4cats
默认缓存模式Cache-Aside，先写 DB 后删缓存
防穿透空值缓存 加 布隆过滤器
防击穿互斥锁 / 逻辑过期 / refreshAfterWrite
防雪崩TTL 随机抖动 加 多级缓存 加 熔断
分布式锁SET NX PX 加 token 加 Lua 释放
限流Lua 原子令牌桶
序列化circe 或 jsoniter，大对象加压缩
key 命名版本前缀 v1:
连接管理Resource 或 ZLayer
监控命中率 加 回源 QPS 加 慢查询

一句话记忆：Scala 缓存 = Caffeine 挡热点（W-TinyLFU、refreshAfterWrite）+ Redis4cats 兜全量（Resource 管理连接）+ Cache-Aside 先写库后删缓存 + 穿透用布隆、击穿用互斥、雪崩用抖动 + key 带版本前缀，把「快」建立在「失效时机想清楚」之上。

延伸阅读

  • /scala-database-access/ — 回源侧的数据库访问与连接池
  • /scala-performance-jvm/ — 本地缓存的 GC 与堆内存开销
  • /scala-configuration-feature-flags/ — TTL 与开关等缓存参数配置
  • /scala-serialization-protocols/ — JSON 与 Protobuf 编解码细节
  • /scala-functional-effects/ — Resource 与效果系统基础
  • /scala-zio-program/ — 用 ZLayer 把缓存服务化
  • /scala-testing-practice/ — Testcontainers 集成测试
  • /scala-observability-logging-tracing/ — 命中率指标与链路追踪
  • Redis 专题 — Redis 数据结构与运维实践
  • 可观测性专题 — 指标采集与告警联动

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「scala」更多文章

  1. Scala 云原生部署实战：容器化、健康检查与 Kubernetes 运维
  2. Scala Web 安全与鉴权实战：JWT、OAuth2 与安全加固
  3. Scala 与 Kafka 集成实战：FS2-Kafka、Alpakka 与流式管道