用 Redis 12+ 特性构建实时协作系统

Redis AI Challenge: 超越缓存

这是 Redis AI Challenge: Beyond the Cache 的参赛作品。

我构建了一个实时协作工作流自动化平台，将 Redis 从简单的缓存转变为完整的应用骨干，名叫 RedisFlow。可以把它想象为 Zapier 与 Figma 的结合体，多个用户可以实时协同设计、执行和监控自动化工作流。

RedisFlow 的独特之处在于：它利用 12+ 个 Redis 特性创建了一个生产级别的平台，展示了 Redis 作为多模型数据库、实时引擎和分布式系统基础的能力——同时实现了 10ms 以下的协作延迟，支持每个工作流 100+ 并发用户。

可视化工作流构建器：包含 7 种节点类型的拖放界面

实时协作：多用户同步编辑，显示实时光标

即时执行：实时状态更新观看工作流运行

实时监控：流式输出执行日志和指标

分布式处理：使用 Redis 驱动的任务队列水平扩展

尝试 RedisFlow：https://redisflow.vercel.app

源代码：https://github.com/depapp/redisflow

RedisFlow 将 Redis 推向了缓存之外，将其用作构建现代实时应用的完整平台。以下是我如何利用每个 Redis 特性的方式：

1. Redis JSON - 工作流主数据库

// 将复杂的工作流定义存储为 JSON 文档
await redisClient.json.set(`workflow:${id}`, '$', {
  id,
  name: 'Customer Data Processor',
  nodes: [...],
  connections: [...],
  metadata: { created: Date.now(), version: 1 }
});

2. Redis Pub/Sub - 实时协作引擎

// 向所有连接用户广播工作流变更
publisher.publish(`workflow:${workflowId}:updates`, JSON.stringify({
  type: 'node-moved',
  nodeId: node.id,
  position: { x: 100, y: 200 },
  userId: currentUser.id
}));

// 订阅接收更新
subscriber.subscribe(`workflow:${workflowId}:updates`);
subscriber.on('message', (channel, message) => {
  // 实时更新 UI
});

3. Redis Streams - 事件溯源与执行日志

// 为实时监控流式输出执行事件
await ioredisClient.xadd(
  `execution:${executionId}:logs`,
  '*',
  'type', 'node_complete',
  'nodeId', node.id,
  'status', 'success',
  'duration', executionTime,
  'result', JSON.stringify(result)
);

4. BullMQ（Redis 驱动）- 分布式任务处理

// 使用重试逻辑将工作流执行加入队列
await executionQueue.add('execute', {
  workflowId,
  executionId,
  inputs
}, {
  attempts: 3,
  backoff: { type: 'exponential', delay: 2000 }
});

5. Redis Search - 全文工作流搜索

// 为工作流创建搜索索引
await redisClient.ft.create('idx:workflows', {
  name: { type: 'TEXT', sortable: true },
  description: 'TEXT',
  tags: 'TAG',
  created: 'NUMERIC'
}, { ON: 'JSON', PREFIX: 'workflow:' });

6. Redis 有序集合 - 指标与排名

// 追踪工作流流行度和执行指标
await ioredisClient.zincrby('workflows:by_executions', 1, workflowId);
await ioredisClient.zadd('workflows:by_date', Date.now(), workflowId);

7. Redis 计数器 - 实时分析

// 原子性地增加执行计数
await ioredisClient.incr('metrics:executions:total');
await ioredisClient.incr(`metrics:executions:daily:${today}`);

8. Redis TTL - 自动数据清理

// 为临时数据设置 TTL
await ioredisClient.setex(`session:${sessionId}`, 3600, userData);
await ioredisClient.expire(`cache:${key}`, 300);

9. Redis 事务 - 原子操作

// 通过多命令事务确保一致性
const multi = ioredisClient.multi();
multi.json.set(`workflow:${id}`, '$', workflowData);
multi.zadd('workflows:by_date', Date.now(), id);
multi.incr('metrics:workflows:total');
await multi.exec();

10. Redis 集合 - 用户在线状态追踪

// 追踪每个工作流的活跃用户
await ioredisClient.sadd(`workflow:${workflowId}:users`, userId);
const activeUsers = await ioredisClient.scard(`workflow:${workflowId}:users`);

11. Redis 列表 - 执行队列管理

// 管理执行顺序
await ioredisClient.lpush('execution:queue', executionId);
const nextExecution = await ioredisClient.rpop('execution:queue');

12. Redis 哈希 - 执行元数据

// 存储执行详情
await ioredisClient.hset(`execution:${executionId}`, {
  status: 'running',
  startedAt: Date.now(),
  workflowId,
  userId
});

架构与性能

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Vue.js    │────▶│   Node.js   │────▶│    Redis    │
│  Frontend   │     │   Backend   │     │   Database  │
└─────────────┘     └─────────────┘     └─────────────┘
       │                    │                    │
       └────WebSocket───────┘                    │
           (Socket.io)                           │
                                                 │
                                          ┌──────┴────┐
                                          │  Features │
                                          ├───────────┤
                                          │ • JSON    │
                                          │ • Streams │
                                          │ • Pub/Sub │
                                          │ • Search  │
                                          │ • BullMQ  │
                                          └───────────┘

工作流加载时间：<50ms（Redis JSON）

协作延迟：<10ms（Redis Pub/Sub）

搜索响应：<100ms（Redis Search）

执行启动：<100ms（BullMQ）

并发用户：每个工作流 100+ 用户

吞吐量：1000+ 次执行/分钟

生产级功能特性

错误处理：具备重试机制的综合错误处理

监控：内置的指标和性能追踪

可扩展性：通过 Redis Cluster 支持水平扩展

安全性：输入验证、限流和用户隔离

持久化：Redis AOF 数据持久化

清晰架构：模块化设计，关注点分离清晰

现代技术栈：Vue 3、Node.js、Socket.io 和 Redis

易于配置：单命令安装，提供环境模板

全面文档：README、API 文档和部署指南

随着 Redis 不断演进，RedisFlow 可以进一步扩展，包括：

向量搜索：AI 驱动的工作流推荐

Redis ML：智能工作流优化

时间序列：高级分析和监控

图数据库：复杂的工作流依赖管理

RedisFlow 证明了 Redis 不仅仅是一个缓存——它是一个强大的多模型平台，能够为整个应用提供动力。通过以创新方式利用 12+ 个 Redis 特性，RedisFlow 创造了用传统数据库难以实现的实时协作体验。

这个项目突破了 Redis 的界限，向全球开发者展示了他们可以使用 Redis 作为主数据库和实时引擎来构建复杂的、真正实时的应用。

如需采取进一步行动，你可以考虑屏蔽此人和/或举报滥用行为