採用 Redis 的批次資料處理防抖機制系統 - n8n 工作流程

開發用於建構高吞吐量資料管線，且需要速率限制功能的開發人員。
需要防止同時執行（Concurrent Executions）衝擊下游服務（例如 API 限制）的 n8n 使用者。
尋找使用 Redis 進行交易管理和並行控制等進階 n8n 範本的工程師。
任何需要在其 n8n 工作流環境中實現穩健核心邏輯流程控制的使用者。

處理資料的突然湧入或快速到達的事件，可能會導致後端系統超載或迅速耗盡 API 速率限制。此 n8n 工作流提供了使用 Redis 來緩解這些事件的穩健解決方案。此 n8n 範本不會對每一個輸入都執行處理邏輯，而是將資料暫存一小段時間。只有在該時間窗內收到資料的『最後一次』執行才被允許繼續處理，並將所有暫存的訊息彙整成一個批次。這能大幅減輕負載、優化資源使用，並展現了 Redis n8n 節點在您的 n8n 環境中進行進階流程控制的強大應用。

這個進階的 n8n 工作流旨在確保與相同 queueid 相關的並行執行能有效且無衝突地被處理。

1. 進入與鎖定檢查（觸發器與 Redis 節點）： n8n 觸發器啟動流程。第一個 n8n 節點會檢查 Redis 上指定的佇列（lock{{queueid}}）是否有啟用中的鎖。如果鎖定啟用，工作流會等待（Wait for lock release）並重試檢查，以防止批次處理同時發生。


2. 緩衝與標記： 如果鎖定未啟用，則將傳入的資料推送至 Redis 清單（messages{{queue_id}}）。Crypto n8n 節點會生成一個唯一的 UUID，並立即將此 ID 寫入『最後更新』的 Redis 鍵中，將當前執行標記為最新的寫入者。


3. 去抖等待： 接著執行會到達定義好的去抖週期（例如 2 秒）的 Wait n8n 節點。這允許後續的並行執行到達並覆蓋『最後更新』的鍵，從而有效地取消當前這次執行。


4. 驗證（我是最後一個嗎？）： 等待期結束後，執行會從 Redis 讀取當前的『最後更新』UUID。Am I last? n8n 節點會檢查讀取到的 UUID 是否與自身生成的 UUID 相符。如果不相符，表示有新的執行已經接管，則當前的 n8n 工作流執行將會安靜終止。


5. 批次處理： 如果執行確認自己是最後的寫入者，它將會繼續獲取佇列鎖、從 Redis 清單中讀取所有訊息（Get messages）、清除該清單、釋放鎖，最後使用 Split messages n8n 節點，將彙整後的批次準備好供後續的處理邏輯使用。

要使用這個強大的 n8n 工作流範本，請遵循以下步驟：

1. 匯入： 複製提供的 JSON，並直接將其作為新的工作流匯入到您的 n8n 實例中。


2. Redis 憑證： 為所有 Redis n8n 節點元件（如 Get lock value、Push to message list 等）設定所需的 Redis 憑證。請確保連線詳細資訊（主機、連接埠、密碼）正確無誤。


3. 觸發器設定： 此 n8n 工作流使用『執行工作流觸發器』（Execute Workflow Trigger）。當您從另一個 n8n 工作流或外部工具執行此工作流時，請務必傳遞兩個必要的參數：queue_id（您正在進行去抖處理的特定資源/佇列的唯一識別碼）和 data（您希望暫存的負載資料）。


4. 去抖時間設定： 調整主要 Wait n8n 節點上的時間（位於驗證步驟之前），以設定所需的去抖時間窗。預設為 2 秒鐘。


5. 下游邏輯： 將您的目標批次處理邏輯連接到 Split messages n8n 節點的輸出端。

觸發器（Execute Workflow Trigger）： 作為 n8n 工作流的進入點，預期接收 queue_id 和 data 參數，以識別並提交訊息到佇列。
Redis 節點（多個實例）： 用於狀態管理的關鍵元件。用於進行去抖 ID 的 get 和 set 等鍵操作、用於將訊息緩衝至清單的 push、用於鎖定的 incr，以及用於清理的 delete。
鎖定是否啟用？(If n8n 節點)： 實作條件邏輯，用以檢查處理佇列目前是否已被鎖定（鎖定值 > 0）。
等待鎖釋放 (Wait n8n 節點)： 暫停 n8n 工作流執行直到鎖被釋放，為並行進入的請求實作了退避（Back-off）邏輯。
等待 (n8n 節點)： 核心的去抖機制。暫停執行固定時長（2 秒），以允許較新的請求到達並重設『最後寫入者』標籤。
加密 (Crypto n8n 節點)： 生成一個唯一的執行識別碼（UUID），作為『最後寫入者』標籤使用。
我是最後一個嗎？(If n8n 節點)： 將 Redis 中當前的『最後寫入者』標籤與自身生成的 UUID 進行比較，以判斷本次執行是否負責處理該批次資料。
拆分訊息 (Split messages n8n 節點)： 接收從 Redis 讀取的彙總訊息清單，並將其拆解回單個資料項，供其他 n8n 節點進行下游處理。