還沒串接 Discord——串上讓片場參與記進 XP，每天還有 3 篇 DE／AI 精選第 3 章：多佇列分流 — 把任務送到專屬的 Worker
Celery 預設所有任務擠在同一條佇列。這一章你會學會依「任務類型」把任務分流到不同佇列（twse / tpex），並讓專門的 worker 各自去處理，互不搶資源。最後再用
--scale一行指令把同一種 worker 開成多份。
做完這一章，你會做到
- 用
.s()+apply_async(queue=...)把不同股票送到不同佇列。 - 啟動兩個「只吃指定佇列」的 worker（
-Q twse、-Q tpex）。 - 確認 twse 的任務只有 twse worker 拿得到，tpex 反之。
- 做一個對照實驗：故意不開某個 worker，看到那條佇列的訊息一直堆積沒人處理。
- 用
--scale把 twse worker 開成 3 份，看一批任務被 3 個實例分擔。 - 講得出「為什麼 worker 可以隨便加減份數」的三個原因。
先搞懂：為什麼要分流
打個比方：一間餐廳只有一台出單機時，熱炒、飲料、甜點的單全混在一起排隊。生意一忙，做甜點的師傅可能被前面一大排熱炒單卡住，明明他很閒卻拿不到自己的單。解法是把出單分成幾條線，各自有專責師傅。
在這個專案裡，你會把股票分成兩類、兩條佇列：
twse佇列：上市股票（例如台積電 2330）tpex佇列：上櫃 / 其他標的（例如美債 ETF 00679B）
分流之後有三個好處：
- 互不干擾：某一類暴量，不會卡住另一類。
- 可獨立擴充：哪條線忙，就只加開那條線的 worker。
- 可差異化設定：不同佇列的 worker 可以各自調併發數、甚至用不同的 pool。
這是第 1 章「多開 worker」的精緻版：不只是「加人」，而是「讓對的人做對的事」。
這一章會用到的檔案
| 檔案 | 角色 | 說明 |
|---|---|---|
crawler/producer_multi_queue_print.py |
生產者（print 版） | 分流 + 只印出，方便你觀察 |
crawler/producer_multi_queue.py |
生產者（正式版） | 分流 + 呼叫會寫 DB 的任務 |
crawler/worker.py |
不動 | 沿用 |
docker-compose-local.yml |
部署 | 定義了 worker_twse / worker_tpex 兩個專屬 worker |
compose-advanced/ |
部署（進階） | 拆開的網路版 compose，--scale 實驗用 |
Dockerfile |
容器化 | worker 容器的 image 定義（--scale 時 build） |
一行一行讀懂 producer（print 版）
from crawler.tasks_crawler_finmind import crawler_finmind_print
# .s() = signature，把「task + 參數」綁成一個可派送物件
task_2330 = crawler_finmind_print.s(stock_id="2330")
# apply_async(queue="twse")：指定這個任務送到 twse 佇列
task_2330.apply_async(queue="twse")
print("send task_2330 task")
task_00679b = crawler_finmind_print.s(stock_id="00679B") # 美債 ETF
task_00679b.apply_async(queue="tpex") # 送到 tpex 佇列
print("send task_00679b task")有三個關鍵你要懂：
.s(stock_id="2330")是什麼？.s是 signature（簽章）的縮寫。它把「要執行哪個 task」加上「帶什麼參數」打包成一個物件，但還沒送出。好處是這個物件可以再附加queue、重試策略等進階設定。你可以想成「填好、但還沒投遞的包裹」。.apply_async(queue="twse")才是真正投遞，而且指定投到名為twse的佇列。- 為什麼不能用
.delay()？ 因為.delay()是簡化版，不能指定佇列。要分流就一定要用完整的apply_async。這就是第 1 章埋的伏筆「想指定佇列就得用 apply_async」在這裡兌現。
正式版
producer_multi_queue.py幾乎一樣，只是把crawler_finmind_print換成會寫資料庫的crawler_finmind。再一次示範「分流邏輯不變，只換任務落地方式」。
Worker 端：用 -Q 只吃指定佇列
專案的 docker-compose-local.yml 裡，兩個 worker 是這樣啟動的（你在本機也可以直接下這兩行）：
# twse 專屬 worker：只消費 twse 佇列
uv run python -m celery -A crawler.worker worker --loglevel=info --hostname=twse@%h -Q twse
# tpex 專屬 worker：只消費 tpex 佇列
uv run python -m celery -A crawler.worker worker --loglevel=info --hostname=tpex@%h -Q tpex兩個新參數：
-Q twse：這個 worker 只處理twse佇列的任務，完全看不到tpex的。這是分流能成立的關鍵——沒加-Q的 worker 只吃預設的celery佇列。--hostname=twse@%h：幫 worker 取個好認的名字（%h會自動換成主機名），這樣你在 Flower / RabbitMQ 上一眼就能分辨誰是誰。
一步一步跟著做
方式 A：本機開兩個 worker（推薦，看得最清楚）
你需要四個終端機：twse worker、tpex worker、producer、瀏覽器。
T1 — 啟動 twse worker：
uv run python -m celery -A crawler.worker worker -l info --hostname=twse@%h -Q twse✅ banner 裡看到
.> queues: twse、worker 名字是twse@...，就過關。
T2 — 啟動 tpex worker：
uv run python -m celery -A crawler.worker worker -l info --hostname=tpex@%h -Q tpex✅ 這個 worker 的 queue 是
tpex。
T3 — 派送分流任務：
uv run crawler/producer_multi_queue_print.py你會看到：
send task_2330 task
send task_00679b task方式 B：用 docker compose 一次起兩個 worker
docker compose -f docker-compose-local.yml up -d --build \
rabbitmq flower mysql phpmyadmin worker_twse worker_tpex第一次要 build image（下載 base image +
uv sync），需要幾分鐘。
確認兩個 worker ready：
docker compose -f docker-compose-local.yml logs worker_twse | grep ready
docker compose -f docker-compose-local.yml logs worker_tpex | grep ready✅ 預期：
twse@xxxx ready.
tpex@xxxx ready.💡 名稱小提醒：這個 compose 的服務名是
worker_twse/worker_tpex（docker compose logs用它），但container_name設成crawler_twse/crawler_tpex（docker ps、docker logs看到的是它）。同一個容器、兩個名字，別以為是兩個東西。
然後用容器裡的 producer 服務派送（或照方式 A 在本機跑 producer）：
docker compose -f docker-compose-local.yml up producer觀察與驗證
觀察點 1：任務只被對應的 worker 拿到
- T1（twse worker） 只會印出 2330 的股價表。
- T2（tpex worker） 只會印出 00679B 的股價表。
- 它們不會拿到對方佇列的任務。
✅ 這是本章最重要的驗證：2330 永遠不會出現在 tpex worker，反過來也一樣。
用 Docker 跑的話，用 log 驗證：
docker compose -f docker-compose-local.yml logs worker_twse | grep succeeded docker compose -f docker-compose-local.yml logs worker_tpex | grep succeeded
觀察點 2：RabbitMQ UI 看到兩條佇列
打開 http://localhost:15672 → Queues and Streams，你會看到 twse、tpex 兩條佇列各自有流量進出。
觀察點 3：Flower 看分工
| 頁籤 | 觀察 |
|---|---|
| Dashboard | 兩個 worker：twse@... 和 tpex@...，都 Online |
| Workers | 點 twse@...，它的 Queues 只有 twse |
| Tasks | 2330 那筆的 worker 欄是 twse，00679B 的 worker 欄是 tpex |
✅ Tasks 頁能清楚看到「哪個任務被哪個 worker 處理」，這是分流生效最直接的證據。
觀察點 4（一定要做的對照實驗）：故意不開某個 worker
- 只開 twse worker，不要開 tpex worker。
- 跑一次
producer_multi_queue_print.py。 - 到 RabbitMQ UI 看
tpex佇列 → Ready 數會是 1，而且一直不動（沒人領）。 - 這時你再把 tpex worker 開起來 → 那則訊息立刻被消化、Ready 歸零。
✅ 這個實驗直接證明了「worker 只吃自己被
-Q指定的佇列」，也順便預告了第 4 章「訊息會在佇列裡等待」的行為。
對照總表
| # | 你應該看到 | 它證明了什麼 |
|---|---|---|
| 1 | twse worker 只印 2330 | -Q 讓 worker 只吃指定佇列 |
| 2 | tpex worker 只印 00679B | 佇列彼此隔離 |
| 3 | RabbitMQ 有 twse / tpex 兩條佇列 | apply_async(queue=) 真的建立了分流 |
| 4 | Flower Tasks 頁 worker 欄對得上 | 分流路徑可以被監控 |
| 5 | 不開 tpex worker → tpex 佇列堆積 | 訊息在佇列等待，不會消失 |
進階：用 --scale 把同一種 worker 開成多份
分流解決了「讓對的人做對的事」。但如果 twse 這條線就是忙不過來呢？加人。 Docker Compose 的 --scale 讓你一行指令把同一個 worker 服務開成 N 份，連程式碼都不用碰。
這一段改用 compose-advanced/ 的網路版 compose（服務拆開、共用一個外部網路），因為 --scale 需要 worker 服務沒有寫死 container_name——名字寫死就沒辦法開第二份。
⚠️ 先把上面整合版停掉，避免 port 衝突：
docker compose -f docker-compose-local.yml down
Step 1：建立共用網路
compose-advanced/ 的檔案都掛在外部網路 my_network 上，要先建：
docker network create my_networkStep 2：啟動 RabbitMQ + Flower、MySQL + phpMyAdmin（網路版）
docker compose -f compose-advanced/rabbitmq.yml up -d
docker compose -f compose-advanced/mysql.yml up -d✅ 驗證：
curl -o /dev/null -s -w "RabbitMQ: %{http_code}\n" http://localhost:15672
curl -o /dev/null -s -w "phpMyAdmin: %{http_code}\n" http://localhost:8080兩個都回 200。
Step 3：用 --scale 啟動 3 個 twse worker
docker compose -f compose-advanced/docker-compose-worker-network.yml up -d --scale crawler_twse=3確認起了幾份：
docker compose -f compose-advanced/docker-compose-worker-network.yml ps✅ 預期：crawler_twse 有 3 個實例（名稱結尾 -1 / -2 / -3），crawler_tpex 有 1 個。
Step 4：發一批 twse 任務，看 3 個實例分擔
uv run python -c "
from crawler.tasks_crawler_finmind import crawler_finmind_print
for sid in ['2330', '2317', '2454', '2412', '1301', '1216']:
crawler_finmind_print.s(stock_id=sid).apply_async(queue='twse')
print('sent', sid)
"然後看 log：
docker compose -f compose-advanced/docker-compose-worker-network.yml logs crawler_twse | tail -40✅ 預期：6 個任務被 3 個不同的實例分擔（log 前綴的容器名不同），整批完成的時間比單一 worker 快。三個實例都在搶同一條 twse 佇列——誰有空誰拿。
Step 5：縮回 1 份、收工
# 縮回 1 份
docker compose -f compose-advanced/docker-compose-worker-network.yml up -d --scale crawler_twse=1
# 全部收掉
docker compose -f compose-advanced/docker-compose-worker-network.yml down
docker compose -f compose-advanced/rabbitmq.yml down
docker compose -f compose-advanced/mysql.yml down
docker network rm my_network💡
mysql.yml down預設保留資料 volume；加-v才會連資料一起刪。
為什麼 worker 可以隨便加減份數？
你剛剛從 1 份 scale 到 3 份、再縮回 1 份，全程沒改任何程式。這不是理所當然的——是三個設計特性讓它成立：
| 特性 | 說明 | 帶來的好處 |
|---|---|---|
| 無狀態（Stateless） | worker 不在本機保存任何狀態，每個任務都是獨立的 | 隨便加減 worker 都不會弄亂資料 |
| 不開對外 port | worker 只「主動連出去」抓 RabbitMQ 的任務，不需要別人連進來 | 不會 port 衝突，才能開多份 |
| 共用 broker | 所有 worker 連同一個 RabbitMQ，從同一條佇列搶任務 | 加 worker = 更多手一起搶，自動負載平衡 |
對照：為什麼 RabbitMQ / MySQL 不能這樣隨便 scale
| Worker | RabbitMQ / MySQL | |
|---|---|---|
| 有狀態？ | 無 | 有（訊息 / 資料） |
| 開對外 port？ | 否 | 是（5672 / 3306） |
| scale 多份？ | ✅ 隨意 | ❌ 需要 cluster 設定 |
記住這句：要 scale 的東西必須無狀態、不搶 port、共用同一個資料/訊息來源。 Celery worker 三者都符合，所以是整套系統裡最容易水平擴充的部分。
想再深入一點
.delay()vs.s().apply_async()：前者簡單但不能指定佇列；後者彈性，能指定queue、countdown（延遲）、eta（指定時間）、link（串接後續任務）等。分流一定要用後者。- 預設佇列叫
celery：沒指定queue的任務會進celery佇列；沒加-Q的 worker 也只吃celery。這解釋了下面那個最常見的 bug。 - 一個 worker 也可以吃多條佇列：
-Q twse,tpex就會同時消費兩條。你可以想想：什麼時候要專職、什麼時候要兼職。
卡住了？常見錯誤這樣排
| 你遇到的狀況 | 原因 | 怎麼解 |
|---|---|---|
| 任務送出去卻永遠沒人做 | worker 沒加對應的 -Q，只吃預設 celery 佇列 |
worker 加 -Q twse（或 tpex） |
想分流卻用了 .delay() |
.delay() 不能指定佇列 |
改用 .s(...).apply_async(queue=...) |
producer 送 twse、worker 卻 -Q twes |
佇列名稱打錯、對不上 | 檢查兩邊佇列名稱完全一致 |
| 兩個 worker 名字一樣互相衝突 | 沒設 --hostname |
各自加不同的 --hostname=xxx@%h |
--scale 報 container name 衝突 |
compose 裡 worker 寫死了 container_name |
用 compose-advanced/ 的網路版（沒寫死名字） |
| scale 的 worker 連不上 rabbitmq | 服務不在同一個 Docker 網路 | 確認都掛在 my_network，且先 docker network create my_network |
想一想（確認你懂了）
先自己想過再看答案。
Q1：如果不分流、全部擠在一條佇列，會有什麼問題？
所有任務混在一起排隊，一種任務暴量時會卡住另一種。例如上市股票的任務突然湧入幾百個，上櫃的任務就得排在後面乾等，即使你其實想讓上櫃優先。而且你沒辦法針對不同類型的任務給不同的資源配置——全部共用同一批 worker、同一個併發設定。
Q2：worker_twse 為什麼看不到 00679B 的任務？
因為 worker_twse 啟動時加了 -Q twse，代表它只訂閱 twse 這條佇列。而 00679B 是被 apply_async(queue="tpex") 送到 tpex 佇列的。worker 只會去拿自己訂閱的佇列裡的訊息，不會去別條佇列撈，所以它根本看不到 00679B。
Q3：假設上市股票的量是上櫃的 10 倍，你會怎麼配置這兩種 worker？
給 twse 更多資源。做法有兩種、通常一起用：多開幾個 -Q twse 的 worker（橫向，--scale crawler_twse=N 一行搞定），以及把 twse worker 的 --concurrency 調高（縱向）；tpex 這邊量小，保守配置就好。這樣資源就配在真正忙的那條線上。
Q4：為什麼分流一定要用 apply_async、不能用 .delay()？
因為 .delay() 是簡化版，它固定把任務送到預設佇列，沒有讓你指定 queue 的參數。要指定佇列就必須用完整的 apply_async(queue=...)（或先用 .s() 綁好再 apply_async）。這也是為什麼多佇列的 producer 都改用 .s().apply_async() 的寫法。
Q5：--scale crawler_twse=3 之後，3 個實例會不會重複做同一個任務？
不會。3 個實例訂閱的是同一條 twse 佇列，而佇列裡的一則訊息一次只會交給一個消費者（RabbitMQ 負責分派）。所以 6 個任務會被 3 個實例「分著做」，不會重工。這也是「共用 broker」帶來的自動負載平衡——不需要你自己寫任何分配邏輯。
換你試試看
練習 1：新增第三條佇列 etf
在 producer 裡加上把 0050、0056 送到 etf 佇列的程式碼：
crawler_finmind_print.s(stock_id="0050").apply_async(queue="etf")
crawler_finmind_print.s(stock_id="0056").apply_async(queue="etf")然後啟動一個只吃 etf 佇列的 worker：
uv run python -m celery -A crawler.worker worker -l info --hostname=etf@%h -Q etf你會看到 0050、0056 只出現在這個 etf worker，twse / tpex worker 都拿不到。這讓你確認：分流的規則完全由「producer 送到哪條佇列」+「worker 訂閱哪條佇列」這兩件事決定。
練習 2：做「不開 worker 就送任務」的對照實驗
只開 twse worker、不開 tpex worker，跑一次 producer，然後去 RabbitMQ UI（http://localhost:15672 → Queues）看 tpex 佇列。
你會看到 tpex 的 Ready 數字停在 1 不動——因為訊息送進佇列了，但沒有任何 worker 訂閱 tpex 去拿它，所以它只能一直等。這時你再把 tpex worker 開起來，那個數字會立刻歸零、任務被消化。Ready 就是「在佇列裡排隊、還沒被拿走」的訊息數，這個實驗讓你看懂佇列的緩衝本質：訊息不會因為沒人處理就消失，它會等。
練習 3：讓一個 worker 兼吃兩條佇列
uv run python -m celery -A crawler.worker worker -l info -n both@%h -Q twse,tpex用這個 worker（不要開別的），跑一次多佇列 producer。你會看到 twse 和 tpex 的任務都被這一個 worker 做掉。這讓你理解 -Q 可以訂閱多條佇列，也讓你想清楚一個設計問題：什麼時候該用「一個 worker 兼多條」（省資源），什麼時候該用「專職 worker」（互不干擾、好擴充）。
練習 4：把 tpex 也 scale 成 2 份
沿用進階段落的環境，跑：
docker compose -f compose-advanced/docker-compose-worker-network.yml up -d --scale crawler_twse=3 --scale crawler_tpex=2
docker compose -f compose-advanced/docker-compose-worker-network.yml ps你會看到兩種 worker 各自 scale 到不同份數。這讓你體會分流 + scale 的組合威力：每一條線都可以獨立決定要投多少人力。
這一章你學到了
- 多佇列 = 依任務類型分流，搭配
-Q讓 worker 各司其職。 .s()+apply_async(queue=...)是分流的關鍵語法，.delay()做不到。--scale一行指令就能把 worker 開成 N 份，靠的是「無狀態、不開 port、共用 broker」三個設計。- 分流（做對的事）× scale（加人手）＝ 分散式系統的資源調度基本功。
下一章要做什麼
到目前為止任務都一帆風順。但真實世界的爬蟲一定會失敗——API 超時、網路斷線、worker 被殺。下一章你會故意讓任務失敗、故意殺掉 worker，觀察訊息在 RabbitMQ 裡的命運，學會 retry、requeue 與 acks_late。
送件鴿