4、Redis高级特性：Pipeline、事务、Lua脚本、Pub/Sub 在行情分发中的应用

行情分发，说白了就是跟时间赛跑。

你想想看，交易所每秒推送几千笔行情，你的系统要在这几毫秒内完成接收、处理、再分发给下游策略。用Redis做这个事，光会用个SET/GET可不够。今天我就把我在实战中打磨出来的几个高级玩法，掰开揉碎了讲给你听。

4.1 Pipeline：把网络开销压到最低

先问个问题：你发一条命令给Redis，等回复，再发下一条，这中间浪费了多少时间？

我实测过，在千兆网络下，一次往返（RTT）大概0.1ms到0.5ms。如果你要批量写入1000条行情数据，光网络等待就占了100ms到500ms。这在交易系统里，简直是灾难。

Pipeline就是解决这个问题的。

它把多条命令攒成一包，一次性发出去，Redis也一次性处理完，再打包回复。网络往返次数从N次降到了1次。

我在项目中遇到过这样一个场景：行情网关需要把逐笔成交数据写入Redis，供多个策略订阅。最开始用逐条写入，CPU没跑满，网络先扛不住了。改成Pipeline后，同样硬件配置，吞吐量从每秒2万笔飙到了15万笔。

// 伪代码示例：使用Pipeline批量写入行情
Pipeline pipeline = jedis.pipelined();
for (MarketData data : batch) {
    // 注意：这里不立即发送，只是缓存命令
    pipeline.set("tick:" + data.getSymbol(), data.toJson());
    pipeline.lpush("recent:" + data.getSymbol(), data.toJson());
    pipeline.ltrim("recent:" + data.getSymbol(), 0, 999); // 只保留最近1000条
}
// 一次性发送并接收所有回复
List<Object> responses = pipeline.syncAndReturnAll();

4.2 事务与Lua脚本：保证原子性

行情分发里有个经典问题：你要同时更新一个股票的当前价和历史最高价，如果中间Redis挂了，数据就不一致了。

Redis事务（MULTI/EXEC）能保证一批命令要么全执行，要么全不执行。但说实话，我很少直接用事务。为什么？因为事务在EXEC之前只是排队，不检查逻辑错误。你写了个语法错误的命令，它照样排队，等到EXEC时才报错。

我更推荐用Lua脚本。

Lua脚本在Redis里是原子执行的，整个脚本执行期间，不会被其他客户端的命令打断。而且脚本里可以做条件判断、循环，比事务灵活得多。

来看一个实战例子：行情分发时，我们需要保证「先更新缓存，再发布通知」这个操作的原子性。

-- Lua脚本：原子更新行情并发布通知
-- KEYS[1] = 行情key, KEYS[2] = 通知channel
-- ARGV[1] = 行情JSON, ARGV[2] = 时间戳

local current = redis.call('GET', KEYS[1])
if current then
    local currentTs = cjson.decode(current).timestamp
    -- 只处理更新的行情（防止旧数据覆盖）
    if tonumber(ARGV[2]) <= tonumber(currentTs) then
        return 0  -- 旧数据，忽略
    end
end

-- 写入新行情
redis.call('SET', KEYS[1], ARGV[1])
-- 发布通知
redis.call('PUBLISH', KEYS[2], ARGV[1])
return 1

这段脚本解决了什么问题？它保证了「判断-写入-发布」三步是原子的。如果不用Lua，在高并发下，两个客户端可能同时读到旧数据，都认为需要更新，结果后一个覆盖了前一个的正确数据。

4.3 Pub/Sub：实时行情分发的最佳拍档

Redis的发布/订阅（Pub/Sub）模式，说白了就是一个消息广播系统。生产者往某个频道发消息，所有订阅了这个频道的消费者都能实时收到。

在行情分发里，这个模式天然匹配：

• 生产者：行情网关，收到交易所数据后往频道里推
• 消费者：各个策略服务，订阅自己感兴趣的股票频道

我设计过一个典型的行情分发架构，用SVG画出来大概是这样：

这个架构的好处很明显：

• 解耦：网关和策略互不感知，新增策略只需订阅对应频道
• 实时：消息推送延迟通常在微秒级
• 灵活：策略可以按股票代码、按板块、按交易所订阅

// 生产者：行情网关推送数据
public void publishTick(TickData tick) {
    String channel = "tick:" + tick.getSymbol();
    jedis.publish(channel, tick.toJson());
}

// 消费者：策略订阅行情
public void subscribeStrategy(Jedis jedis, String... symbols) {
    for (String symbol : symbols) {
        String channel = "tick:" + symbol;
        jedis.subscribe(new JedisPubSub() {
            @Override
            public void onMessage(String channel, String message) {
                TickData tick = TickData.fromJson(message);
                // 处理行情...
                strategy.onTick(tick);
            }
        }, channel);
    }
}

4.4 组合拳：Pipeline + Lua + Pub/Sub 实战

单个特性用起来不难，难的是把它们组合成一套高效的行情分发链路。我分享一个我在实盘环境中验证过的方案：

环节	使用技术	解决的问题
行情接收	Pipeline	批量写入行情数据，降低网络开销
数据校验	Lua脚本	原子判断行情时序，防止旧数据覆盖
实时分发	Pub/Sub	毫秒级推送到所有订阅策略
异常恢复	List + BRPOP	断连后从持久化队列补数据

具体流程是这样的：

1. 行情网关收到交易所数据，用Pipeline批量写入Redis
2. 写入时调用Lua脚本，校验时间戳，保证数据一致性
3. 脚本执行成功后，通过PUBLISH命令推送到对应频道
4. 策略服务订阅频道，实时接收行情
5. 同时，网关把原始数据写入List，作为持久化备份

最后说一句，Redis这些高级特性，单独看都不复杂。但真正考验功力的，是怎么根据业务场景把它们组合起来。行情分发只是其中一个例子，你可以在自己的系统里举一反三。

---

公众号:蓝海数据掘金营，微信deep3321