本文是专栏「n8n 工作流引擎剖析」第 05 章组件深度剖析的第 05/11 篇承接上一篇《工作流运行器 Workflow Runner》。核心是ScalingService、JobProcessor、worker命令。这是一个可选子系统常规模式下不存在。你在这里读完本文你能解释为什么队列里只携带一个执行 ID、一个工作进程如何把一个任务变成一次运行以及结果和 HTTP 响应是如何传回主进程的。缩写DBDatabase数据库、IDIdentifier标识符、HTTPHypertext Transfer Protocol超文本传输协议、JSONJavaScript Object NotationJavaScript 对象表示法、TTLTime To Live存活时间、MCPModel Context Protocol模型上下文协议只出现在任务字段里。角色回顾负责把执行 ID从主进程/webhook 进程搬运到工作进程基于 Redis 的 Bull并把进度和结果转发回去。掌握队列名称、任务优先级、工作进程并发数、Leader 选举。不做不在任务里携带运行数据不自己运行节点那是工作进程里的引擎做的。出现于S6入队和S7工作进程启动以及 S12 末尾的返回路径。内部设计进程角色工作进程(n8n worker --concurrency10)主进程 / webhook 进程(n8n start, n8n webhook)1 创建执行记录2 addJob({executionId,…})3 投递4 加载执行记录 工作流5 进度 最终数据job.progress(msg)6 global:progress工作流运行器ScalingService生产者 监听器RedisBull 队列ScalingService.setupWorkerqueue.process(concurrency)JobProcessor.processJobWorkflowExecute数据库execution_entity execution_data图注Redis 双向都只携带小小的任务消息所有运行数据都走数据库。任务负载——刻意做得很小JobDatacli/src/scaling/scaling.types.tsworkflowId、executionId、loadStaticData、pushRef?、streamingEnabled?、restartExecutionId?、projectId?、poolName?、encryptedRunnerIdentity?外加几个 MCP 专用字段。**没有节点没有数据项没有凭据。**为什么运行数据可能很大Redis 不是存放它的好地方。S6 中创建的那条数据库记录已经包含了完整的IRunExecutionData工作进程直接读它就行。一个小任务丢了重试的代价很低。生产者一侧 ——enqueueExecutionworkflow-runner.ts第 740 行惰性加载ScalingService并调用setupQueue()创建一个maxStalledCount: 0的 Bull 队列以及 Redis 客户端scaling.service.ts第 97–107 行。poolConfigService.resolvePoolForExecution(data)选出{ queueName, poolName }工作进程池。scalingService.addJob(jobData, { priority: realtime ? 50 : 100, queueName })。通过scaling-main钩子getLifecycleHooksForScalingMain运行workflowExecuteBefore并返回一个PCancelablea取消时调用stopJob(job)b任务结果到达时 resolve。消费者一侧 ——JobProcessor.processJobjob-processor.ts第 121 行executionPersistence.findSingleExecution(executionId, { includeData: true, unflattenData: true })找不到 → 报错。如果状态是crashed→什么也不做防止 Bull 隐式重试机制把一条已崩溃的执行再跑一遍。executionRepository.setRunning(executionId)→ 状态变为running写入startedAt。可选地从工作流记录重新加载staticDataloadStaticData。计算执行超时从execution.workflowData构建new Workflow(...)getBase(...)生成additionalData。构建工作进程钩子getLifecycleHooksForScalingWorker加上两个经 Bull 转发的处理器sendResponse→job.progress({ kind: respond-to-webhook, executionId, response, workerId })sendChunk→job.progress({ kind: send-chunk, … })new WorkflowExecute(additionalData, execution.mode, execution.data).processRunExecutionData(workflow)如果没有executionData——即手动/部分运行——则改用manualExecutionService.runManually。结束时发送job-finishedv2 版本带有status、startedAt、stoppedAt、waitTill、lastNodeExecuted、metadata。返回路径——global:progress上的消息JobMessage的种类有respond-to-webhook、job-finished、job-failed、abort-job、send-chunk、mcp-response。主进程的registerMainOrWebhookListenersscaling.service.ts会根据msg.kind做分发种类主进程的动作respond-to-webhook解码响应体activeExecutions.resolveResponsePromise(executionId, response)→ 正在等待的 HTTP 调用方收到回复send-chunkactiveExecutions.sendChunk→ 流式发给调用方job-finished解析任何挂起的响应存入jobResults[executionId]之所以需要是因为removeOnComplete: true会丢弃任务原本的返回值job-failed把错误连同堆栈一起记录日志工作进程监听abort-job消息并调用jobProcessor.stopJob(jobId)。多主进程与 Leader 选举leader-election-client.ts、multi-main-setup.ee.ts、redis-lock.service.ts当有多个主进程时其中一个是Leader只有它负责管理非 webhook 类触发器触发器注册中心和定期恢复。像add-webhooks-triggers-and-pollers这类命令走的是 n8n 自己的发布/订阅通道不是Bull——JobMessage的注释特别提醒不要把两者搞混。交互关系对象契约工作流运行器addJob、stopJob、popJobResult任务优先级 50实时vs. 100数据库双方都读写执行记录工作进程的钩子负责持久化结果引擎工作进程用存储下来的IRunExecutionData实例化WorkflowExecuteRedisBull 队列键名带有queue.bull.prefix前缀⚓ 回到示例 —— S6 → S7主进程添加{workflowId:wf123,executionId:1042,loadStaticData:false,projectId:ada 所在项目,poolName:解析出的池}取值仅为示意字段名是真实的JobData字段。某个工作进程的queue.process(JOB_TYPE_NAME, concurrency, processJob)收到它。日志info 级别Worker started execution 1042 (job 77)processJob读取执行1042状态new数据里的栈是[Order Webhook]把它设为running构建Workflow接好工作进程钩子调用processRunExecutionData。因为Order Webhook用的是onReceived模式不需要sendResponse消息——调用方已经在主进程上拿到回复了。如果 Ada 选的是lastNode模式Respond的流程会是工作进程sendResponse钩子 →job.progress(respond-to-webhook)→ Redis → 主进程的监听器 →resolveResponsePromise→ 主进程上那个仍然打开着的 HTTP 连接。失败行为运行器插入记录工作进程 setRunning引擎顺利跑完节点报错 / job-failedstopJob/abort-job/超时Wait 节点设置 waitTill恢复webhook / 定时器进程挂了恢复逻辑标记恢复逻辑newrunningsuccesserrorcanceledwaitingcrashed图注跨进程的执行状态生命周期crashed是由恢复逻辑写入的不是由这次运行本身写入的。失败情形行为工作进程在运行中途挂掉数据库中的执行会停留在running恢复逻辑会把处于进行中状态的记录new、running、unknown——合称CRASHABLE_EXECUTION_STATUSES标记为crashed。maxStalledCount: 0能阻止 Bull 悄悄地重新投递一个已停滞的任务Bull 重试一条已崩溃的执行如果状态已经是crashedprocessJob返回{ success: false }入队时 Redis 不可用走错误路径执行以error结束并重新抛出异常工作进程启动时连不上 Redis致命错误Error initializing Lua scripts→ 工作进程process.exit(1)从界面取消主进程stopJob(job)→abort-job消息 → 工作进程中止那个PCancelable工作进程还在跑时主进程挂了工作进程仍会跑完并写入数据库丢失的是回复响应 Promise执行记录不会丢下一篇《执行引擎 Execution EngineWorkflowExecute》n8n 的心脏决定下一个该跑哪个节点、用什么数据。 返回专栏目录
` 到安全可靠的文本处理)
