我正在開發一個新的監控系統，該系統可以測量 Celery 隊列吞吐量，并在隊列備份時幫助向團隊發出警報。在我的工作過程中，我遇到了一些我不理解的奇怪行為（并且在 Celery 規范中沒有詳細記錄）。出于測試目的，我設置了一個端點，該端點將使用 16 個可用于模擬備份隊列的長時間運行的任務填充隊列。框架是Flask，隊列代理是Redis。Celery 配置為每個工作人員可以并行處理最多 4 個任務，而我有 2 個工作人員正在運行。api/health.pydef health(): health = Blueprint("health", __name__) @health.route("/api/debug/create-long-queue", methods=["GET"]) def long_queue(): for i in range(16): sleepy_job.delay() return make_response({}, 200) return [email protected](priority=HIGH_PRIORITY)def sleepy_job(*args, **kwargs): time.sleep(30)以下是我模擬備份生產隊列的方法：我打電話/api/debug/create-long-queue來模擬隊列中的備份。根據上面的計算，每個工人應該忙著睡覺 1 分鐘（總共可以同時處理 8 個任務。每個任務只休眠 30 秒，總共 16 個任務。）不久后（< 5 秒）我進行了另一個 API 調用，這會啟動具有實際業務邏輯的不同工作（處理入站 Webhook API 調用）。我們稱這個工作為handle_incoming_message。這是我看到的使用花來檢查隊列的內容：雖然所有工作線程都被前 8 個任務阻塞，但我在隊列中看 sleepy_job不到新任務的跡象 ，盡管我確信第二個 API 調用已被調用。handle_incoming_messagehandle_incoming_message.delay()前 8 個sleepy_job任務完成后（約 30 秒），我 handle_incoming_message在隊列中看到新任務，狀態為RECIEVED。在第二個（也是最后一個）8 個sleepy_job任務完成后，我現在看到handle_incoming_message了狀態STARTED（當 UI 使用該任務中接收和處理的新數據進行更新時，我可以確認這一點。）問題因此，很明顯，當工作人員在處理完前 8 個sleepy_job任務后暫時解除阻塞時，他們正在以一種可見的方式做一些事情來標記/確認新任務。但這留下了幾個懸而未決的問題：handle_incoming_messagehandle_incoming_message當worker被阻塞時，新任務的狀態如何？工作人員解除封鎖后發生了哪些變化，使得它現在可以了解新handle_incoming_message任務？“已接收”狀態實際上意味著什么？（獎勵：如何在工作人員被阻塞時了解排隊的任務？）
1 回答
紅糖糍粑
TA貢獻1815條經驗 獲得超6個贊
當所有工作人員都被阻止時，某些任務可能由于預取而處于已接收狀態（請參閱文檔）。因此，您的任務很可能只是在隊列中，等待 Celery 工作進程（協調進程 - 這些不是實際的工作進程）接收。
Flower 是一項簡單的服務，它基于稱為“任務事件”的 Celery 功能構建。簡單來說，它（Flower）將自己訂閱為所有事件（已接收、成功、啟動、失敗等）的接收者，然后以可視方式將這些事件呈現給 Web 客戶端。因此，當 Celery 工作線程收到任務時，就會發送“任務已收到”事件。Flower 獲取此事件，并更改儀表板中該任務的狀態。
當一個任務被“接收”時，這意味著特定的 Celery 工作線程將該任務從隊列中取出，并且它可以立即執行（如果有空閑的工作進程來執行它），或者 Celery 工作線程將等待工作進程成為準備運行任務。我已經提到過預取 - Celery 工作進程通常會執行比可用工作進程更多的任務。
Celery 沒有為用戶提供列出特定隊列中的內容的方法。
添加回答
舉報
0/150
提交
取消