Flowise 的异步队列：BullMQ + Redis 的事件驱动架构

学习目标

• 理解 Flowise 三种队列的职责与触发场景
• 掌握 BaseQueue 抽象类的统一接口设计
• 了解 Redis 事件分发与分布式限流器的实现

前置知识

本章为项目特定的工程实践。建议了解：

• 消息队列的基本概念（生产者、消费者、Job）
• Redis 的基本数据结构与 Pub/Sub 机制

---

项目实践

三种队列类型

队列	职责	触发场景
PredictionQueue	异步执行 Flow 预测	大量并发预测请求
UpsertQueue	批量文档插入向量库	大批量文档 Upsert
ScheduleQueue	定时触发 Flow	定时任务（如每天 9 点执行某个 Flow）

BaseQueue 抽象类

所有队列继承 BaseQueue 抽象类：

abstract class BaseQueue {
    protected queue: Queue           // BullMQ Queue
    protected queueEvents: QueueEvents
    protected connection: RedisOptions
    private worker: Worker

    abstract processJob(data: any): Promise<any>
    abstract getQueueName(): string
    abstract getQueue(): Queue

    public async addJob(jobData: any): Promise<Job> {
        return await this.queue.add(jobId, jobData, {
            removeOnFail: keepJobConfig,
            removeOnComplete: keepJobConfig
        })
    }

    public createWorker(concurrency = 100000): Worker {
        this.worker = new Worker(this.queue.name, async (job) => {
            return await this.processJob(job.data)
        })
    }
}

关键设计：

• addJob 统一 Job 添加接口，自动处理去重和清理
• createWorker 统一 Worker 创建接口，可配置并发数
• Job 完成后自动清理（可配置保留策略：REMOVE_ON_AGE、REMOVE_ON_COUNT）

Redis 事件分发

多实例部署下，SSE 事件需要跨实例分发：

RedisEventPublisher：将 SSE 事件发布到 Redis Streams

RedisEventSubscriber：订阅 Redis Streams，将事件推送到本地的 SSE 连接

RateLimiterManager 分布式限流

Flowise 使用 express-rate-limit + RedisStore 实现分布式限流：

class RateLimiterManager {
    private rateLimiters: Map<string, RateLimitRequestHandler>

    addChatflowRateLimiter(chatflowId, limitMax, limitDuration, limitMsg) {
        const store = new RedisStore({
            sendCommand: (cmd) => this.redisClient[cmd]()
        })
        this.rateLimiters.set(chatflowId, rateLimit({
            windowMs: limitDuration,
            max: limitMax,
            message: limitMsg,
            store
        }))
    }
}

动态限流更新（队列模式下）：

this.queueEvents.on('updateRateLimiter', async ({ limitDuration, limitMax, limitMsg, id }) => {
    this.addChatflowRateLimiter(id, limitMax, limitDuration, limitMsg)
})

---

问题与规避

Redis 单点故障

问题：BullMQ 依赖 Redis，Redis 宕机导致所有队列无法工作。

对策：

• 使用 Redis Sentinel 或 Redis Cluster 提高可用性
• 配置 REDIS_TLS 和 REDIS_KEEP_ALIVE 保持连接稳定
• 在队列模式下，如果 Redis 不可用，Fallback 到同步执行模式

事件顺序保证

问题：Redis Streams 的事件可能被不同实例以不同顺序消费，导致 SSE 事件乱序。

对策：

• BullMQ 的 streams: { events: { maxLen: QUEUE_REDIS_EVENT_STREAM_MAX_LEN } } 配置限制 Stream 长度，防止事件堆积
• 每个 Job 的事件通过 job.log 保持顺序
• 前端 SSE 客户端应能处理乱序事件（通过消息 ID 重排序）

Worker 并发与资源竞争

问题：默认 WORKER_CONCURRENCY = 100000，可能导致 LLM API 被大量并发请求打满。

对策：

• 根据 LLM API 的速率限制配置合理的并发数
• 使用 RateLimiterManager 为每个 Chatflow 设置独立限流
• 监控 LLM API 的 429 错误，自动调整并发

---

设计取舍

BullMQ vs 其他队列方案

方案	优势	代价
BullMQ	Redis-based、TypeScript 原生、事件系统完善	依赖 Redis
RabbitMQ	更成熟的协议、支持多种 Exchange 模式	部署复杂
SQS	托管服务、零运维	AWS 绑定、延迟较高
内存队列	零依赖、最快	进程重启丢失、不支持多实例

Flowise 选择 BullMQ 的原因：

• Redis 已经是 Flowise 的缓存后端，无额外依赖
• TypeScript 原生支持，与 Flowise 的技术栈一致
• 事件系统（QueueEvents）天然支持 SSE 推送

同步执行 vs 队列模式

维度	同步执行	队列模式
延迟	低（直接执行）	高（入队 + 出队）
并发	受限于单实例	多实例水平扩展
可靠性	进程宕机丢失	Redis 持久化
适用场景	低流量、单实例	高流量、多实例

Flowise 支持两种模式：默认同步执行（低延迟），高流量场景启用队列模式。

---

参考来源

• BullMQ 官方文档
• Redis Streams 文档