Cloud Run functions 入門：函式會在什麼時候執行？

前言

如果只是收到一個請求或事件後做件小事，未必需要先架好一整個服務。Cloud Run functions（舊稱 Cloud Functions）讓你部署函式，交由平台在觸發條件成立時執行；但觸發來源、權限、逾時和費用仍要自己設定與確認。

這篇先比較第一代與第二代，再用幾種觸發方式看函式會收到什麼資料、失敗時怎麼處理。最後整理選型和費用上的注意事項。

什麼是 Cloud Run functions？

Cloud Run functions 讓你部署單一函式來回應事件：

flowchart TB
  subgraph INTAKE["1. 接收與啟動"]
    direction TB
    SOURCE["HTTP／Storage／Pub/Sub<br/>或其他 Eventarc 事件"] --> ROUTE["Trigger Routing<br/>篩選並標準化"]
    ROUTE --> CAPACITY{"有可用的<br/>函式 Instance？"}
    CAPACITY -->|"沒有"| START["啟動 Instance<br/>載入 runtime 與程式碼"]
    CAPACITY -->|"有"| INVOKE["呼叫 entry point"]
    START --> INVOKE
  end
  subgraph EXECUTION["2. 執行與確認"]
    direction TB
    WORK["執行業務邏輯<br/>存取 Secret、DB 或 API"] --> RESULT{"執行結果"}
    RESULT -->|"成功"| ACK["回應 HTTP<br/>或確認事件"]
    RESULT -->|"可重試失敗"| RETRY["來源依政策重送<br/>同一事件可能再次出現"]
  end
  subgraph SCALING["3. 閒置與縮容"]
    direction TB
    IDLE{"持續一段時間<br/>沒有新流量？"}
    IDLE -->|"是，min = 0"| ZERO["縮容至零<br/>停止實例計費"]
    IDLE -->|"否或保留 min"| WARM["維持可用 Instance<br/>等待後續事件"]
  end
  INVOKE --> WORK
  ACK --> IDLE

最後那一步有個正式名稱叫 scale-to-zero（縮容至零）——沒有流量時實例會全部關掉，這段時間 Google 一毛錢都不跟你收。代價是下一個請求進來時得重新冷啟動（後面會講），這是無伺服器服務最迷人也最容易卡住新手的特性。

2024 年 Google 將 Cloud Functions Gen2 更名為 Cloud Run functions，因為 Gen2 底層就是跑在 Cloud Run 上。但 gcloud 指令和介面仍然沿用 functions 名稱——你後面看到的指令還是 gcloud functions deploy，別被名字搞混了。

Gen1 vs Gen2

Gen2 是目前預設且推薦的版本，Gen1 是功能較受限的原始版本（官方仍持續支援）。

表格超出欄寬時，可左右捲動；鍵盤可聚焦表格後用左右方向鍵查看。

記憶體與 CPU 上限這兩列：函式配額頁列的函式記憶體上限是第 1 代 8GiB、現行版 32GiB，Cloud Run 的記憶體設定頁也寫最高 32 GiB（要搭配 8 vCPU；4 vCPU 最多 16 GiB）。版本比較頁的 Instance size 則寫現行版「Up to 16 GiB RAM with 4 vCPU」。兩頁官方寫法不同，要用到 16 GiB 以上時，部署前先在主控台確認（2026-10-12 查證）。

並發處理——Gen2 的最大優勢

Gen1 每個實例只能處理一個請求。100 個同時請求 = 啟動 100 個實例。

Gen2 支援並發（主控台建立時預設 80；用 gcloud functions deploy 沒帶 --concurrency 時預設是 1），設成 80 的話 100 個同時請求只需要 2 個實例：

flowchart TB
  BURST["同一時間收到<br/>100 個請求"] --> VERSION{"執行環境"}
  VERSION -->|"Gen1<br/>concurrency = 1"| GEN1["最多啟動 100 個<br/>Instance<br/>每個只處理 1 個請求"]
  VERSION -->|"Gen2<br/>concurrency = 80"| GEN2["約啟動 2 個 Instance<br/>每個可同時處理多個請求"]
  GEN1 --> COST1["冷啟動與實例數較多<br/>共享狀態風險較低"]
  GEN2 --> SAFETY{"程式是否<br/>thread-safe，<br/>且不把使用者<br/>資料放全域變數？"}
  SAFETY -->|"否"| FIX["移除共享可變狀態、<br/>限制並發<br/>或改用外部儲存"]
  SAFETY -->|"是"| COST2["較少 Instance 與冷啟動<br/>再依 CPU／記憶體壓測調整"]
  FIX --> COST2

這裡的冷啟動（cold start）指的是：實例從零被叫醒、載入你的程式碼、初始化執行環境的那段延遲，通常幾百毫秒到幾秒。實例越少、被重複利用的機會越高，冷啟動就越少——這也是 Gen2 並發省成本之外的另一個好處。

💡 重點提點

Gen2 並發是把雙面刃。同一個實例同時處理多個請求，意思是這些請求共用同一份記憶體——全域變數、資料庫連線、暫存檔，全部是共享的。新手最常踩的坑是：把使用者資料存進全域變數，結果 A 的請求讀到 B 的資料。寫並發函式時，請假設任何兩個請求隨時可能同時跑，狀態要嘛放區域變數、要嘛寫進外部儲存（Firestore、Memorystore），別偷懶塞全域。

觸發器類型

下面會用到幾個 GCP 服務名詞，先一句話交代：GCS（Cloud Storage）是 Google 的物件儲存，放圖片、檔案、備份；Pub/Sub 是訊息佇列，把「發送方」跟「處理方」解耦；Firestore 是 NoSQL 文件資料庫。觸發器的角色就是：當這些服務發生事情（檔案上傳、收到訊息、文件變更），自動叫醒你的函式。

HTTP 觸發器

最直接的一種，函式本身就是一個 HTTPS Endpoint（端點，也就是一個可以被呼叫的網址）：

# Python 範例
import functions_framework

@functions_framework.http
def hello(request):
    name = request.args.get("name", "World")
    return f"Hello, {name}!"

# 部署
gcloud functions deploy hello \
  --gen2 \
  --runtime=python312 \
  --trigger-http \
  --allow-unauthenticated \
  --region=asia-east1 \
  --entry-point=hello

# 取得 URL
gcloud functions describe hello \
  --region=asia-east1 \
  --format="value(serviceConfig.uri)"

Cloud Storage 觸發器

GCS 有檔案變動時自動執行：

import functions_framework
from cloudevents.http import CloudEvent

@functions_framework.cloud_event
def on_file_upload(cloud_event: CloudEvent):
    data = cloud_event.data
    bucket = data["bucket"]
    name = data["name"]
    print(f"New file: gs://{bucket}/{name}")

gcloud functions deploy on-file-upload \
  --gen2 \
  --runtime=python312 \
  --trigger-event-filters="type=google.cloud.storage.object.v1.finalized" \
  --trigger-event-filters="bucket=my-upload-bucket" \
  --region=asia-east1 \
  --entry-point=on_file_upload

Pub/Sub 觸發器

收到 Pub/Sub 訊息時自動執行：

import base64
import functions_framework
from cloudevents.http import CloudEvent

@functions_framework.cloud_event
def on_message(cloud_event: CloudEvent):
    data = base64.b64decode(cloud_event.data["message"]["data"]).decode()
    print(f"Received message: {data}")

gcloud functions deploy on-message \
  --gen2 \
  --runtime=python312 \
  --trigger-topic=my-topic \
  --region=asia-east1 \
  --entry-point=on_message

Eventarc 觸發器（Gen2 獨有）

Eventarc 是 GCP 的統一事件路由服務——它把各種 Google Cloud 服務（甚至第三方）發生的事件，標準化後轉送給你的函式。有了它，Gen2 不用為每種來源各寫一套整合，支援 90+ 種事件來源，包括：

• Cloud SQL 的管理操作（走 Cloud Audit Logs，不是資料列的異動）
• BigQuery 作業完成
• Cloud Build 的建置操作（走 Cloud Audit Logs，例如建立、取消建置的 API 呼叫）
• Firestore 文件變更
• 第三方服務（Datadog、Check Point CloudGuard 等，Preview）

flowchart TB
  START["什麼事情要叫醒函式？"] --> DIRECT{"呼叫方是否能直接送出<br/>同步 HTTP／Webhook 請求？"}
  DIRECT -->|"是"| HTTP["HTTP Trigger<br/>自行設計認證、回應與逾時"]
  DIRECT -->|"否，是雲端事件"| SOURCE{"事件來源"}
  SOURCE -->|"Cloud Storage 物件變更"| GCS["Cloud Storage Trigger<br/>用 bucket 與 event type 篩選"]
  SOURCE -->|"應用程式發布訊息"| PUBSUB["Pub/Sub Trigger<br/>解耦生產者與消費者"]
  SOURCE -->|"Firestore 或其他 Google Cloud 事件"| EVENTARC["Eventarc Trigger<br/>以 CloudEvents 統一路由"]
  SOURCE -->|"第三方事件"| THIRD["先接 Webhook 或發布到 Pub/Sub<br/>再轉成可重試事件"]
  HTTP --> CONTRACT["定義 payload、IAM、重試、<br/>dead-letter 與可觀測性"]
  GCS --> CONTRACT
  PUBSUB --> CONTRACT
  EVENTARC --> CONTRACT
  THIRD --> CONTRACT

支援的語言

表格超出欄寬時，可左右捲動；鍵盤可聚焦表格後用左右方向鍵查看。

各語言的支援版本會持續往前推進（新版本上架、舊版本退役都很頻繁），實際部署前以官方的 runtime support 頁面為準。備考時不必背某個 runtime 的精確小版本，記得「Cloud Run functions 支援主流語言、版本跟著官方滾動」這個概念就夠了。

資源配置與冷啟動

資源配置

# 記憶體上限 32Gi、vCPU 上限 8（32Gi 要配 8 vCPU）
# min-instances 消除冷啟動、max-instances 控制成本
# concurrency 是每實例並發數（Gen2）、timeout 單位是秒
gcloud functions deploy my-func \
  --gen2 \
  --runtime=nodejs24 \
  --trigger-http \
  --memory=512Mi \
  --cpu=1 \
  --min-instances=1 \
  --max-instances=20 \
  --concurrency=80 \
  --timeout=60 \
  --region=asia-east1

消除冷啟動

# 保留暖機實例
gcloud functions deploy my-func \
  --min-instances=1

# 設定 0 就是允許縮容至零（預設）
gcloud functions deploy my-func \
  --min-instances=0

flowchart TB
  subgraph COLD["1. 冷啟動取捨"]
    direction TB
    START["量測單次請求資源、<br/>執行時間與冷啟動"] --> LATENCY{"能接受冷啟動？"}
    LATENCY -->|"可以"| MIN0["min-instances = 0<br/>節省閒置成本"]
    LATENCY -->|"不可以"| MIN1["min-instances ≥ 1<br/>保留暖機容量"]
  end
  subgraph PARALLEL["2. 單實例並發"]
    direction TB
    SAFE{"程式 thread-safe，<br/>單請求資源量穩定？"}
    SAFE -->|"是"| CONCURRENCY["逐步提高 concurrency<br/>用壓測找甜蜜點"]
    SAFE -->|"否"| SERIAL["降低 concurrency<br/>先修正共享狀態"]
  end
  subgraph GUARDRAIL["3. 下游與成本護欄"]
    direction TB
    DOWNSTREAM{"Database／API 可承受<br/>最大平行量？"}
    DOWNSTREAM -->|"有限制"| MAX["max × concurrency<br/>限制最壞同時量"]
    DOWNSTREAM -->|"容量充足"| MAX2["仍設定合理 max<br/>避免無上限成本"]
    MAX --> TIMEOUT["timeout 保留重試<br/>與清理時間"]
    MAX2 --> TIMEOUT
  end
  MIN0 --> SAFE
  MIN1 --> SAFE
  CONCURRENCY --> DOWNSTREAM
  SERIAL --> DOWNSTREAM

事件交付語義

At-Least-Once（至少一次）

Cloud Run functions 的事件觸發是 at-least-once（至少一次） 語義，意思是同一個事件可能觸發你的函式好幾次——Google 保證「至少送到一次」，但不保證「剛好一次」。

💡 重點提點

事件可能重送，所以函式要能安全處理同一個事件多次。Event ID 可用來辨認重複事件，但「先查是否處理過，再執行」若沒有交易或唯一鍵保護，仍會遇到並行競態。扣款、發信和訂單寫入還需要各自的冪等設計；下方範例先展示概念，再說明正式環境要補的保護。

解法：設計冪等函式

所謂冪等（idempotent），白話講就是「同一個操作做幾次，效果都跟做一次一樣」。下面用事件的唯一 ID 先查一次「處理過了沒」，處理過就直接跳出：

from google.cloud import firestore

db = firestore.Client()

@functions_framework.cloud_event
def process_order(cloud_event):
    event_id = cloud_event["id"]  # 每個事件有唯一 ID

    # 檢查是否已經處理過
    doc = db.collection("processed_events").document(event_id).get()
    if doc.exists:
        print(f"Already processed: {event_id}")
        return

    # 執行業務邏輯
    do_business_logic(cloud_event.data)

    # 記錄已處理
    db.collection("processed_events").document(event_id).set({
        "processed_at": firestore.SERVER_TIMESTAMP
    })

上面的「先查、執行、再寫」適合解釋概念，但正式環境不能把三步當成天然原子操作；兩個重複事件同時通過查詢時，仍可能一起執行副作用。內部資料更新要用交易或唯一鍵保護，外部 API 則要傳遞相同的 idempotency key，或用 Outbox／工作佇列把狀態與副作用拆開。

flowchart TB
  EVENT["收到 CloudEvent<br/>讀取 Event ID"] --> CLAIM{"以交易／唯一鍵取得<br/>唯一處理權？"}
  CLAIM -->|"否，已完成"| DUPLICATE["重複交付<br/>直接確認"]
  CLAIM -->|"否，處理中或失敗"| RECOVER["依狀態與 lease<br/>等待、接手或重試"]
  CLAIM -->|"是"| KIND{"副作用類型"}
  KIND -->|"內部資料"| TX["在交易中寫入資料<br/>並標記 completed"]
  KIND -->|"外部 API"| KEY{"API 支援<br/>idempotency key？"}
  KEY -->|"是"| API["以 Event ID 當 key 呼叫<br/>成功後標記 completed"]
  KEY -->|"否"| OUTBOX["原子寫入 Outbox<br/>由 worker 提交並更新狀態"]

Secret Manager 整合

方式一：部署時引用（推薦）

# 掛載為環境變數
gcloud functions deploy my-func \
  --set-secrets=DB_PASSWORD=db-password:latest

# 掛載為檔案
gcloud functions deploy my-func \
  --set-secrets=/secrets/api-key=api-key:latest

方式二：程式碼中呼叫

from google.cloud import secretmanager

def get_secret(secret_id):
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/my-project/secrets/{secret_id}/versions/latest"
    response = client.access_secret_version(name=name)
    return response.payload.data.decode("utf-8")

IAM 需求：函式的 Service Account（服務帳號，函式執行時代表的身分）需要 roles/secretmanager.secretAccessor 這個 IAM 角色，否則讀 Secret 會被拒。權限不足是新手部署完第一次踩的雷，看到 PermissionDenied 先回來檢查這一條。

本地開發與測試

Cloud Run functions 提供 Functions Framework，可在本地模擬雲端執行環境：

安裝與啟動

# Python
pip install functions-framework
functions-framework --target=hello --debug --port=8080

# Node.js
npm install @google-cloud/functions-framework
npx functions-framework --target=hello --port=8080

測試 HTTP 函式

curl http://localhost:8080?name=Bobo

測試事件驅動函式（CloudEvents）

curl -X POST http://localhost:8080 \
  -H "ce-specversion: 1.0" \
  -H "ce-type: google.cloud.storage.object.v1.finalized" \
  -H "ce-source: //storage.googleapis.com/projects/my-project" \
  -H "ce-id: test-001" \
  -H "Content-Type: application/json" \
  -d '{"bucket": "my-bucket", "name": "test.txt"}'

定價與免費層

⚠️ 官方定價頁已把 Gen2（Cloud Run functions）併入 Cloud Run 計費，下面是 Gen2 的額度。Gen1 用的是不同單位、不同數字：200 萬次呼叫、400,000 GB-秒、200,000 GHz-秒、5 GB 對外流量——兩代別搞混。

免費層（每月，Gen2）

表格超出欄寬時，可左右捲動；鍵盤可聚焦表格後用左右方向鍵查看。

超過免費層

表格超出欄寬時，可左右捲動；鍵盤可聚焦表格後用左右方向鍵查看。

定價就是 Cloud Run 定價，因為 Gen2 底層直接跑在 Cloud Run 上。

Cloud Run functions vs Cloud Run vs App Engine

三者的差異整理如下：

表格超出欄寬時，可左右捲動；鍵盤可聚焦表格後用左右方向鍵查看。

選型公式

flowchart TB
  START["先描述要部署的單位<br/>與觸發方式"] --> EVENT{"是否由單一事件觸發<br/>一段聚焦邏輯？"}
  EVENT -->|"是"| FUNCTIONS["Cloud Run functions<br/>函式 + HTTP／事件 Trigger"]
  EVENT -->|"否"| CONTAINER{"是否已有容器，或需要任意 runtime、<br/>多路由與完整服務控制？"}
  CONTAINER -->|"是"| RUN["Cloud Run Service<br/>部署容器化 HTTP／gRPC 服務"]
  CONTAINER -->|"否"| WEB{"是否為受支援 runtime 的<br/>傳統 Web 應用？"}
  WEB -->|"是"| APPENGINE["App Engine<br/>平台化 Web App 與版本管理"]
  WEB -->|"否，有開始與完成的長任務"| JOBS["Cloud Run Jobs<br/>批次、ETL 或排程任務"]
  FUNCTIONS --> DONE["以逾時、事件來源、並發、<br/>網路與維運需求再驗證"]
  RUN --> DONE
  APPENGINE --> DONE
  JOBS --> DONE

ACE 考試重點整理

必背知識點

1. Gen2 是預設，底層是 Cloud Run，Gen1 是功能較受限的原始版本
2. Gen2 支援並發（最高 1000/實例），Gen1 只能 1 個請求/實例
3. HTTP 觸發最長 60 分鐘（Gen2），事件觸發最長 9 分鐘
4. 事件交付是 at-least-once，需要冪等設計
5. Eventarc 讓 Gen2 支援 90+ 種事件來源
6. 免費層：每月 200 萬次呼叫
7. Functions Framework 可本地測試

常見陷阱題

Q：Cloud Run functions 能處理長時間運行的 ETL 任務嗎？ A：HTTP 觸發最長 60 分鐘，事件觸發最長 9 分鐘。超過這個時間就該改用 Cloud Run Jobs（最長 7 天）。

Q：100 個用戶同時呼叫 Gen1 函式，需要幾個實例？ A：100 個。Gen1 每實例只能處理 1 個請求。Gen2 只需要 2 個實例（concurrency=80）。

Q：Cloud Run functions 保證事件只處理一次嗎？ A：不保證。Cloud Run functions 是 at-least-once，需要自己設計冪等邏輯。

Q：GCS 上傳自動觸發處理，用什麼服務？ A：Cloud Run functions（GCS 觸發器最自然的搭配）。

實戰範例：圖片上傳自動生成縮圖

# main.py
import functions_framework
from cloudevents.http import CloudEvent
from google.cloud import storage
from PIL import Image
import io

storage_client = storage.Client()

@functions_framework.cloud_event
def generate_thumbnail(cloud_event: CloudEvent):
    data = cloud_event.data
    bucket_name = data["bucket"]
    file_name = data["name"]

    # 跳過非圖片和已是縮圖的檔案
    if not file_name.lower().endswith((".jpg", ".jpeg", ".png")):
        return
    if file_name.startswith("thumbnails/"):
        return

    # 下載原圖
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)
    image_data = blob.download_as_bytes()

    # 生成縮圖
    image = Image.open(io.BytesIO(image_data))
    image.thumbnail((200, 200))

    # 上傳縮圖
    output = io.BytesIO()
    image.save(output, format="JPEG", quality=85)
    output.seek(0)

    thumb_blob = bucket.blob(f"thumbnails/{file_name}")
    thumb_blob.upload_from_file(output, content_type="image/jpeg")
    print(f"Thumbnail created: thumbnails/{file_name}")

# 部署
gcloud functions deploy generate-thumbnail \
  --gen2 \
  --runtime=python312 \
  --trigger-event-filters="type=google.cloud.storage.object.v1.finalized" \
  --trigger-event-filters="bucket=my-image-bucket" \
  --region=asia-east1 \
  --memory=512Mi \
  --entry-point=generate_thumbnail

flowchart TB
  subgraph EVENT["1. 事件進入"]
    direction TB
    UPLOAD["上傳原始檔案"] --> FINALIZED["Storage 發出<br/>object finalized"]
    FINALIZED --> FUNCTION["Eventarc 呼叫<br/>generate_thumbnail"]
  end
  subgraph GUARDS["2. 前置條件"]
    direction TB
    PREFIX{"路徑以<br/>thumbnails/ 開頭？"}
    PREFIX -->|"是"| STOP1["函式自己的輸出<br/>直接結束"]
    PREFIX -->|"否"| TYPE{"是 JPG／JPEG／PNG？"}
    TYPE -->|"否"| STOP2["非圖片<br/>直接結束"]
    TYPE -->|"是"| PROCESS["下載、縮放、編碼"]
  end
  subgraph OUTPUT["3. 寫回與終止"]
    direction TB
    WRITE["寫入 thumbnails/ 路徑"] --> AGAIN["新物件再次產生<br/>finalized 事件"]
    AGAIN --> LOOPSTOP["下一次呼叫命中<br/>prefix guard<br/>不再產生新物件"]
  end
  FUNCTION --> PREFIX
  PROCESS --> WRITE

總結

Cloud Run functions 適合把範圍明確的 HTTP 或事件處理邏輯交給平台執行。上線前要確認觸發來源、逾時與並行設定；事件可能重送，會修改資料的函式也要處理重複執行。這些設定比只記產品名稱更容易影響實際結果。

若需求逐漸變成完整的 API 或持續運作的服務，再比較 Cloud Run service；已有 App Engine 應用，則把遷移成本與現有部署流程一併納入選擇。

下一課 ACE-212：App Engine 怎麼部署和切流量？先選 Standard 或 Flexible，我們來看 GCP 這個歷史最悠久的 PaaS 平台，今天還適合用在哪些場景。

運算服務選型系列

表格超出欄寬時，可左右捲動；鍵盤可聚焦表格後用左右方向鍵查看。