微服务架构将单体应用拆分为独立部署的服务单元，每个服务拥有自己的数据域和生命周期。在这一架构下，Redis 不再只是简单的缓存组件，而是演变为支撑服务间通信、数据共享、流量治理和高可用保障的基础设施。本文系统梳理 Redis 在微服务场景中的八大核心模式，并提供可落地的代码实现。
一、四种经典缓存模式
缓存与持久化存储的协作方式是微服务数据层的基石。根据读写职责的分配位置不同，业界演化出四种经典模式。
1.1 Cache-Aside（旁路缓存）
Cache-Aside 由应用层显式管理缓存与数据库的双写操作，Redis 本身不感知数据库的存在。读取流程：先查缓存，命中则直接返回；未命中则查数据库，写入缓存后再返回。写入流程：先更新数据库，再删除或更新缓存。
这种模式实现简单、控制粒度细，是微服务中最常用的方案。缺点是应用代码中充斥着缓存操作逻辑。
Python 示例（使用 redis-py）：
import redis
import json
from typing import Optional
r = redis.Redis(host='localhost', port=6379, db=0)
def get_user(user_id: str) -> Optional[dict]:
cache_key = f"user:{user_id}"
cached = r.get(cache_key)
if cached:
return json.loads(cached)
user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
if user:
r.setex(cache_key, 300, json.dumps(user))
return user
def update_user(user_id: str, data: dict) -> None:
db.execute(f"UPDATE users SET ... WHERE id = {user_id}")
r.delete(f"user:{user_id}")
1.2 Read-Through（通读缓存）
Read-Through 将缓存管理逻辑下沉到缓存层自身。当缓存未命中时，由缓存组件自动从数据库加载数据并回填。应用只需与缓存交互，无需感知数据库的存在。
这种模式适合读多写少、数据结构稳定的场景，但需要缓存层具备数据源接入能力。
Java 示例（基于 Redisson 的 RMap 配合 MapLoader）：
import org.redisson.Redisson;
import org.redisson.api.*;
import org.redisson.api.map.MapLoader;
public class ReadThroughCache {
public static void main(String[] args) {
RedissonClient redisson = Redisson.create();
RMap<String, User> userMap = redisson.getMap("user:cache",
MapOptions.<String, User>defaults()
.loader(new MapLoader<>() {
@Override
public User load(String key) {
return userDao.findById(key);
}
@Override
public Iterable<String> loadAllKeys() {
return List.of();
}
}));
User user = userMap.get("u1001");
}
}
1.3 Write-Through（通写缓存）
Write-Through 要求数据写入时同步更新缓存和数据库，两者处于强一致状态。适用于对一致性要求极高的场景，如金融账户余额。缺点在于写操作延迟较高，受数据库 IO 瓶颈制约。
1.4 Write-Behind（异步回写）
Write-Behind 先将数据写入缓存并立即返回成功，再由后台异步线程批量回写数据库。这种方式写性能最优，但存在短暂的数据不一致窗口，适合对最终一致性可容忍的计数、日志类场景。
Go 示例（使用 go-redis + 后台 goroutine 回写）：
package main
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/redis/go-redis/v9"
)
type WriteBehindCache struct {
rdb *redis.Client
ctx context.Context
}
func (wbc *WriteBehindCache) Write(key string, value map[string]any) {
data, _ := json.Marshal(value)
wbc.rdb.Set(wbc.ctx, key, data, 0)
go func() {
time.Sleep(5 * time.Second)
dbWrite(key, value)
}()
}
二、多级缓存（本地 L1 + Redis L2）
微服务中 Redis 虽然延迟极低，但相较本地内存仍有数量级的差距。构建多级缓存可以在保证一定一致性的前提下，将热点数据的读取延迟降至微秒级。
2.1 架构设计
请求 -> Caffeine/Guava (L1) -> Redis (L2) -> 数据库
L1 本地缓存使用进程内数据结构（如 Go 的 bigcache、Java 的 Caffeine、Python 的 cachetools），容量小、延迟最低。L2 Redis 作为全局共享缓存，保证多实例间的基础一致性。
2.2 一致性策略
多级缓存的最大挑战在于跨层级的一致性。推荐采用以下策略组合：
- 写入时失效 L1 + 更新 L2：写操作完成后，向所有服务实例广播 L1 失效消息
- 消息驱动同步：借助 Redis Pub/Sub 或消息队列发布缓存变更事件
- 版本号/时间戳标记：缓存值携带版本号，读取时做版本校验，避免脏读
Java 多级缓存示例：
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Cache;
import redis.clients.jedis.Jedis;
public class MultiLevelCache {
private final Cache<String, Object> localCache = Caffeine.newBuilder()
.maximumSize(10000)
.expireAfterWrite(60, TimeUnit.SECONDS)
.build();
private final Jedis jedis = new Jedis("localhost", 6379);
public Object get(String key) {
Object value = localCache.getIfPresent(key);
if (value != null) return value;
String redisValue = jedis.get(key);
if (redisValue != null) {
localCache.put(key, redisValue);
return redisValue;
}
Object dbValue = loadFromDatabase(key);
if (dbValue != null) {
jedis.setex(key, 300, serialize(dbValue));
localCache.put(key, dbValue);
}
return dbValue;
}
public void invalidate(String key) {
localCache.invalidate(key);
jedis.del(key);
jedis.publish("cache:invalidate", key);
}
}
三、缓存一致性方案
微服务环境下，缓存一致性跨越了进程边界和网络拓扑。常见方案包括 TTL 自然过期、主动失效和消息驱动三种。
3.1 TTL 自然过期
最简单的一致性策略，依赖 Redis 的键过期机制自动清理缓存。优点是实现零成本，缺点是过期窗口内存在数据陈旧问题。
3.2 主动失效
数据变更时主动删除或更新缓存。关键在于先更新数据库，后删除缓存（Cache-Aside），避免并发场景下的脏写。对于高一致性要求的场景，可采用延迟双删：删除缓存 -> 更新数据库 -> 延迟再次删除缓存。
3.3 消息驱动一致性
借助 Redis Pub/Sub 或消息队列实现跨服务的缓存失效广播。某服务更新数据后，发布变更事件，所有订阅方收到消息后清除本地缓存。
Python 消息驱动示例：
import redis
import threading
r = redis.Redis()
def invalidate_listener():
pubsub = r.pubsub()
pubsub.subscribe('cache:invalidate')
for message in pubsub.listen():
if message['type'] == 'message':
key = message['data'].decode()
local_cache.pop(key, None)
threading.Thread(target=invalidate_listener, daemon=True).start()
def update_with_broadcast(key, value):
db.update(key, value)
r.delete(key)
r.publish('cache:invalidate', key)
四、CQRS：Redis 作为读模型
CQRS（Command Query Responsibility Segregation）将命令（写）与查询（读）分离。在微服务中，写操作写入关系型数据库保证事务性，读操作由 Redis 承担，实现高性能的查询侧。
4.1 架构流程
写命令 -> 业务服务 -> MySQL -> CDC/Binlog -> 数据同步服务 -> Redis 读模型
查询请求 -> API 网关 -> 查询服务 -> Redis 读模型
4.2 数据同步方式
- 双写模式：服务层写入 MySQL 后同步写入 Redis。简单直接，但存在分布式事务风险
- CDC 模式：通过 Canal、Debezium 等工具捕获 MySQL binlog，异步将变更同步至 Redis。解耦写与读，一致性由事件保证
- 定时同步：定时任务全量或增量同步。适合对实时性要求不高的报表类场景
Go 实现示例（基于 Canal 的 binlog 监听器）：
type BinlogHandler struct {
rdb *redis.Client
ctx context.Context
}
func (h *BinlogHandler) OnRow(e *canal.RowsEvent) error {
if e.Table.Name == "orders" && e.Action == canal.UpdateAction {
for i := 0; i < len(e.Rows); i += 2 {
newRow := e.Rows[i+1]
orderId := newRow[0].(int64)
status := newRow[3].(string)
key := fmt.Sprintf("order:%d", orderId)
h.rdb.HSet(h.ctx, key, "status", status)
}
}
return nil
}
五、分布式限流
微服务网关层必须对流量进行管控，防止突发流量压垮后端服务。Redis 因其原子性和低延迟，成为分布式限流器的最佳载体。
5.1 令牌桶（Token Bucket）
令牌桶以固定速率向桶中放入令牌，请求需消耗令牌才能通过。突发流量可在桶容量允许范围内被平滑处理。
Python 令牌桶实现：
import redis
import time
r = redis.Redis()
def token_bucket(key: str, capacity: int, rate: float) -> bool:
pipe = r.pipeline()
now = time.time()
pipe.hmget(key, ['tokens', 'last_time'])
result = pipe.execute()
tokens = float(result[0][0]) if result[0][0] else capacity
last_time = float(result[0][1]) if result[0][1] else now
elapsed = now - last_time
tokens = min(capacity, tokens + elapsed * rate)
if tokens >= 1:
pipe.hmset(key, {'tokens': tokens - 1, 'last_time': now})
pipe.expire(key, 60)
pipe.execute()
return True
else:
pipe.hmset(key, {'tokens': tokens, 'last_time': now})
pipe.expire(key, 60)
pipe.execute()
return False
5.2 滑动窗口
滑动窗口将时间划分为多个小窗口，统计每个窗口内的请求数。比固定窗口更平滑，能避免窗口边界突发问题。
Java 滑动窗口实现（Redis + Lua 保证原子性）：
String luaScript =
"local key = KEYS[1] " +
"local window = tonumber(ARGV[1]) " +
"local limit = tonumber(ARGV[2]) " +
"local now = tonumber(ARGV[3]) " +
"redis.call('ZREMRANGEBYSCORE', key, 0, now - window) " +
"local current = redis.call('ZCARD', key) " +
"if current < limit then " +
" redis.call('ZADD', key, now, now) " +
" redis.call('EXPIRE', key, window) " +
" return 1 " +
"else " +
" return 0 " +
"end";
boolean allow(String key, long windowMs, int limit) {
long now = System.currentTimeMillis();
Long result = (Long) jedis.eval(luaScript,
Collections.singletonList(key),
Arrays.asList(String.valueOf(windowMs),
String.valueOf(limit),
String.valueOf(now)));
return result == 1;
}
5.3 漏桶（Leaky Bucket）
漏桶以固定速率处理请求，超出容量的请求被拒绝。适合严格限制平均速率的场景，如 API 配额管理。
Go 漏桶实现：
func leakyBucket(rdb *redis.Client, key string, capacity int, leakRate float64) bool {
ctx := context.Background()
script := `
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local leakRate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local state = redis.call('HMGET', key, 'water', 'last_time')
local water = tonumber(state[1]) or 0
local last = tonumber(state[2]) or now
water = math.max(0, water - (now - last) * leakRate)
if water + 1 <= capacity then
water = water + 1
redis.call('HMSET', key, 'water', water, 'last_time', now)
redis.call('EXPIRE', key, 60)
return 1
else
return 0
end
`
res, err := rdb.Eval(ctx, script, []string{key}, capacity, leakRate, time.Now().Unix()).Result()
return err == nil && res.(int64) == 1
}
六、微服务会话共享
传统单体应用使用服务器内存存储会话，这在微服务架构下行不通：请求可能被负载均衡到任意实例，会话数据必须在服务间共享。
6.1 Redis 会话方案
将用户会话序列化后存储至 Redis，所有服务实例连接同一 Redis 集群读取会话。具备以下优势：
- 无状态服务：任意服务实例宕机不影响会话
- 弹性伸缩：新增或缩减实例无需迁移会话
- 跨域共享：不同子域名下的服务可共享同一会话
- 过期管理：利用 Redis TTL 自动清理过期会话
6.2 安全实践
- 会话 ID 使用加密安全的随机字符串
- 敏感字段（如密码哈希）不存入会话
- 设置合理的过期时间并支持滑动续期
- 登录态变更时重置会话 ID 防止固定会话攻击
Python Flask 会话示例：
from flask import Flask, session
import redis
import uuid
import json
r = redis.Redis()
app = Flask(__name__)
@app.route('/login')
def login():
user_id = authenticate(request)
sid = str(uuid.uuid4())
r.setex(f"session:{sid}", 3600, json.dumps({
'user_id': user_id,
'login_at': time.time()
}))
session['sid'] = sid
return {'status': 'ok'}
@app.route('/profile')
def profile():
sid = session.get('sid')
if not sid:
return {'error': 'unauthorized'}, 401
data = r.get(f"session:{sid}")
if not data:
return {'error': 'session expired'}, 401
user = json.loads(data)
r.expire(f"session:{sid}", 3600)
return {'user_id': user['user_id']}
七、服务发现与注册
微服务之间需要动态感知彼此的存在。Redis 可以作为轻量级的服务发现后端，尤其适用于中小规模的微服务集群。
7.1 基于 Redis 的服务注册
每个服务实例启动时向 Redis 注册自身信息，并设置带 TTL 的键。使用心跳机制续期，实例宕机后 TTL 到期自动剔除。
Go 服务注册示例：
type ServiceInstance struct {
ID string `json:"id"`
Name string `json:"name"`
Address string `json:"address"`
Port int `json:"port"`
Metadata map[string]string `json:"metadata"`
}
func (s *ServiceInstance) Register(rdb *redis.Client, ttl time.Duration) {
ctx := context.Background()
key := fmt.Sprintf("service:%s:%s", s.Name, s.ID)
data, _ := json.Marshal(s)
rdb.Set(ctx, key, data, ttl)
go func() {
ticker := time.NewTicker(ttl / 2)
defer ticker.Stop()
for range ticker.C {
rdb.Expire(ctx, key, ttl)
}
}()
}
func Discover(rdb *redis.Client, serviceName string) []*ServiceInstance {
ctx := context.Background()
keys, _ := rdb.Keys(ctx, fmt.Sprintf("service:%s:*", serviceName)).Result()
var instances []*ServiceInstance
for _, key := range keys {
data, err := rdb.Get(ctx, key).Result()
if err != nil {
continue
}
var s ServiceInstance
json.Unmarshal([]byte(data), &s)
instances = append(instances, &s)
}
return instances
}
7.2 健康检查与负载均衡
消费者定期从 Redis 拉取可用实例列表，结合负载均衡策略（如 Round Robin、一致性哈希）进行请求分发。Redis 的过期机制天然实现了服务的健康检查功能，无需额外的健康检查服务。
八、熔断降级集成
微服务调用链中，某个下游服务的故障可能引发级联雪崩。熔断器在检测到异常比例超标后打开熔断，快速失败避免资源耗尽。Redis 可存储熔断器状态和统计指标，实现分布式熔断策略共享。
8.1 基于 Redis 的熔断器状态存储
各服务实例将请求成功/失败计数存入 Redis，所有实例共享同一熔断状态。当失败率超过阈值时，所有实例同时进入熔断状态。
Python 熔断器实现：
import redis
import time
class RedisCircuitBreaker:
def __init__(self, rdb, key, failure_threshold=5, recovery_timeout=30):
self.rdb = rdb
self.key = key
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
def call(self, func, *args, **kwargs):
state = self.rdb.hgetall(f"cb:{self.key}")
status = state.get(b'status', b'closed').decode()
if status == 'open':
last_failure = float(state.get(b'last_failure', 0))
if time.time() - last_failure < self.recovery_timeout:
raise Exception("Circuit breaker is OPEN")
self.rdb.hset(f"cb:{self.key}", 'status', 'half-open')
try:
result = func(*args, **kwargs)
self._record_success()
return result
except Exception as e:
self._record_failure()
raise e
def _record_failure(self):
pipe = self.rdb.pipeline()
pipe.hincrby(f"cb:{self.key}", 'failures', 1)
pipe.hset(f"cb:{self.key}", 'last_failure', time.time())
pipe.execute()
failures = int(self.rdb.hget(f"cb:{self.key}", 'failures') or 0)
if failures >= self.failure_threshold:
self.rdb.hset(f"cb:{self.key}", 'status', 'open')
def _record_success(self):
self.rdb.hdel(f"cb:{self.key}", 'failures')
self.rdb.hset(f"cb:{self.key}", 'status', 'closed')
8.2 降级数据存储
熔断打开期间，服务可返回 Redis 中预置的降级数据或缓存的兜底结果，保证核心功能的可用性。例如电商系统中，推荐服务熔断时返回 Redis 中的热门商品列表，而非空结果。
总结
Redis 在微服务架构中的角色已从单纯的缓存扩展为分布式系统的基础设施。从四种缓存模式的设计，到多级缓存的性能优化；从 CQRS 读写分离，到分布式限流与会话共享；从服务发现到底层熔断降级，Redis 以出色的性能和丰富的数据结构支撑了微服务的各个治理维度。
在实际落地中，应根据业务特点选择合适的模式组合：
- 读多写少：Cache-Aside + 多级缓存
- 写密集 + 最终一致性：Write-Behind
- 高一致读场景：Read-Through / Write-Through
- 流量入口：令牌桶或滑动窗口限流
- 用户态管理：Redis 会话 + Token 续期
- 服务治理：Redis 服务发现 + 状态共享熔断器
掌握这些模式，能够将 Redis 从"会用"提升到"用好"的层次，在微服务架构中构建出高性能、高可用、可扩展的数据层基础设施。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。