3. Redis 高级特性：发布订阅、事务、Lua脚本与Pipeline批量操作

在热点削峰与审批流数据传输场景中，Redis 的基础数据结构（String、List、Hash、Set、Sorted Set）提供了数据存储与访问能力，但面对高并发、强一致性、复杂逻辑和批量处理需求时，必须借助其高级特性来构建可靠的企业级解决方案。本章深入剖析发布订阅、事务、Lua 脚本和 Pipeline 四大核心特性，并结合审批流场景给出最佳实践。

3.1 发布订阅（Pub/Sub）：异步解耦与实时通知

发布订阅模式实现了消息生产者（Publisher）与消费者（Subscriber）之间的完全解耦。在审批流中，当某个审批节点状态变更时，系统可通过发布订阅机制实时通知下游服务或前端，无需轮询数据库。

核心概念与命令

审批流实时通知示例

// 发布者：审批状态变更时通知
PUBLISH approval:flow:12345 '{"action":"approve","userId":"u1001","timestamp":1700000000}'

// 订阅者：监听特定审批流
SUBSCRIBE approval:flow:12345

// 模式订阅：监听所有审批流事件
PSUBSCRIBE approval:flow:*

3.2 事务（Transaction）：原子性与乐观锁

Redis 事务通过 MULTI、EXEC、DISCARD、WATCH 命令实现。它保证一个事务内的所有命令要么全部执行，要么全部不执行（原子性），但不支持传统数据库的回滚机制。

事务执行流程

1. 开启事务： MULTI 标记事务开始。
2. 命令入队： 后续命令进入队列，不立即执行。
3. 执行事务： EXEC 按顺序执行队列中所有命令。
4. 取消事务： DISCARD 清空队列并退出事务。

WATCH 乐观锁：防止并发冲突

在审批流中，多个节点可能同时尝试更新同一审批单状态。使用 WATCH 监控 key，若在事务执行前 key 被其他客户端修改，事务将失败并返回 nil。

// 伪代码：乐观锁更新审批状态
WATCH approval:status:12345
currentStatus = GET approval:status:12345
if (currentStatus == "PENDING") {
    MULTI
    SET approval:status:12345 "APPROVED"
    SET approval:operator:12345 "u1001"
    EXEC
} else {
    UNWATCH
    // 返回冲突提示
}

3.3 Lua 脚本：复杂逻辑的原子性保障

Lua 脚本是 Redis 中实现复杂、多步骤原子操作的最佳方式。脚本在服务端执行，天然具备原子性，且可减少网络往返次数。在热点削峰和审批流中，Lua 脚本常用于实现限流、库存扣减、状态机流转等场景。

核心优势

审批流状态机 Lua 脚本示例

-- 审批状态流转脚本
-- KEYS[1]: 审批单状态 key
-- ARGV[1]: 当前状态
-- ARGV[2]: 目标状态
-- ARGV[3]: 操作人

local current = redis.call('GET', KEYS[1])
if current == ARGV[1] then
    redis.call('SET', KEYS[1], ARGV[2])
    redis.call('SET', KEYS[1] .. ':operator', ARGV[3])
    return 1  -- 成功
else
    return 0  -- 状态冲突
end

// Java 客户端调用示例（Jedis）
String script = "local current = redis.call('GET', KEYS[1]) ..."; // 上述脚本
String sha = jedis.scriptLoad(script);
Object result = jedis.evalsha(sha, 
    Arrays.asList("approval:status:12345"), 
    Arrays.asList("PENDING", "APPROVED", "u1001"));

3.4 Pipeline 批量操作：极致吞吐优化

Pipeline 允许客户端将多条命令一次性发送到 Redis 服务端，服务端处理完毕后一次性返回结果。在热点削峰场景中，Pipeline 能显著降低网络往返时间（RTT），提升批量数据写入或读取的吞吐量。

Pipeline vs 普通批量操作

审批流批量状态更新示例

// Java 示例：使用 Pipeline 批量更新审批状态
Pipeline pipeline = jedis.pipelined();
for (String approvalId : batchApprovalIds) {
    pipeline.set("approval:status:" + approvalId, "APPROVED");
    pipeline.set("approval:time:" + approvalId, String.valueOf(System.currentTimeMillis()));
}
List<Object> results = pipeline.syncAndReturnAll();

3.5 四大特性对比与选型指南