Cloud RunとCloud Pub/SubのPullサブスクリプションで非同期処理を試してみた

export PROJECT_ID=YOUR_PROJECT_ID
export REGION=asia-northeast1
export TOPIC_ID=order-events
export SUBSCRIPTION_ID=order-sub
export PRODUCER_SA=order-api-sa
export CONSUMER_SA=order-worker-sa

# トピックを作成
gcloud pubsub topics create $TOPIC_ID --project=$PROJECT_ID

# Pull サブスクリプションを作成
gcloud pubsub subscriptions create $SUBSCRIPTION_ID \
  --topic=$TOPIC_ID \
  --project=$PROJECT_ID

# 確認
gcloud pubsub topics list --project=$PROJECT_ID
gcloud pubsub subscriptions list --project=$PROJECT_ID

# トピック
user@hostname:~$ gcloud pubsub topics list --project=$PROJECT_ID
---
name: projects/your-project-id/topics/order-events

# Pull サブスクリプション
user@hostname:~$ gcloud pubsub subscriptions list --project=$PROJECT_ID
---
ackDeadlineSeconds: 10
expirationPolicy:
  ttl: 2678400s
messageRetentionDuration: 604800s
name: projects/your-project-id/subscriptions/order-sub
pushConfig: {}
state: ACTIVE
topic: projects/your-project-id/topics/order-events

# プロデューサー用 SA（Pub/Sub へのパブリッシュ権限）
gcloud iam service-accounts create $PRODUCER_SA \
  --display-name="Order API Service Account" \
  --project=$PROJECT_ID

gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="serviceAccount:$PRODUCER_SA@$PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/pubsub.publisher"

# コンシューマー用 SA（サブスクリプションからの Pull 権限）
gcloud iam service-accounts create $CONSUMER_SA \
  --display-name="Order Worker Service Account" \
  --project=$PROJECT_ID

gcloud pubsub subscriptions add-iam-policy-binding $SUBSCRIPTION_ID \
  --member="serviceAccount:$CONSUMER_SA@$PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/pubsub.subscriber" \
  --project=$PROJECT_ID

# プロデューサー用 SA（Pub/Sub へのパブリッシュ権限）
user@hostname:~$   gcloud projects get-iam-policy $PROJECT_ID \
    --flatten="bindings[].members" \
    --filter="bindings.members=serviceAccount:$PRODUCER_SA@$PROJECT_ID.iam.gserviceaccount.com" \
    --format="table(bindings.role)"
ROLE
roles/pubsub.publisher

# コンシューマー用 SA（サブスクリプションからの Pull 権限）
user@hostname:~$ gcloud pubsub subscriptions get-iam-policy $SUBSCRIPTION_ID \
> --project=$PROJECT_ID
bindings:
- members:
  - serviceAccount:order-worker-sa@your-project-id.iam.gserviceaccount.com
  role: roles/pubsub.subscriber
etag: BwZSIhRXtDo=
version: 1

order-worker/
├── worker.py
└── Dockerfile

import os
import time
from google.cloud import pubsub_v1

PROJECT_ID = os.environ.get("PROJECT_ID")
SUBSCRIPTION_ID = os.environ.get("SUBSCRIPTION_ID")
subscription_path = f"projects/{PROJECT_ID}/subscriptions/{SUBSCRIPTION_ID}"

print(f"Worker starting. Watching {subscription_path}...")

subscriber = pubsub_v1.SubscriberClient()

def callback(message):
    try:
        data = message.data.decode("utf-8")
        print(f"Processing: {data}")
        time.sleep(5)  # 注文処理を模したスリープ
        print(f"Done: {data}")
        message.ack()
    except Exception as e:
        print(f"Error: {e}")
        message.nack()

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"Listening for messages on {subscription_path}...")

with subscriber:
    try:
        streaming_pull_future.result()
    except Exception as e:
        print(f"Streaming pull failed: {e}")

PROJECT_ID = os.environ.get("PROJECT_ID")
SUBSCRIPTION_ID = os.environ.get("SUBSCRIPTION_ID")
subscription_path = f"projects/{PROJECT_ID}/subscriptions/{SUBSCRIPTION_ID}"

subscriber = pubsub_v1.SubscriberClient()

def callback(message):
    try:
        data = message.data.decode("utf-8")  # バイト列を文字列に変換
        ...
        message.ack()   # 正常完了 → Pub/Sub にメッセージ削除を通知
    except Exception as e:
        message.nack()  # 処理失敗 → Pub/Sub に再配信を要求

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)

with subscriber:
    streaming_pull_future.result()  # ここでブロックし続ける

FROM python:3.12-slim
RUN pip install google-cloud-pubsub
COPY worker.py .
CMD ["python", "-u", "worker.py"]

cd order-worker

gcloud run worker-pools deploy order-worker \
  --source . \
  --region $REGION \
  --service-account="$CONSUMER_SA@$PROJECT_ID.iam.gserviceaccount.com" \
  --instances=1 \
  --set-env-vars PROJECT_ID=$PROJECT_ID,SUBSCRIPTION_ID=$SUBSCRIPTION_ID

Deploying from source requires an Artifact Registry Docker repository to store built containers. A repository named 
[cloud-run-source-deploy] in region [asia-northeast1] will be created.

Do you want to continue (Y/n)?  Y

Building using Dockerfile and deploying container to Cloud Run worker pool [order-worker] in project [your-project-id] region [asia-northeast1]
✓ Deploying new worker pool... Building Container.                                                                                 
  ✓ Creating Container Repository...                                                                                               
  ✓ Uploading sources...                                                                                                           
  ✓ Building Container... Logs are available at [ https://console.cloud.google.com/cloud-build/builds;region=asia-northeast1/YOUR_BUILD_ID?project=YOUR_PROJECT_NUMBER ].                                                                           
  . Creating Revision...                                                                                                           
Done.                                                                                                                              
Worker pool [order-worker] revision [order-worker-00001-zz6] has been deployed.
Done.

See logs with:
gcloud beta run worker-pools logs tail order-worker

Or visit https://console.cloud.google.com/run/worker-pools/details/asia-northeast1/order-worker

order-api/
├── main.py
└── requirements.txt

import os
from flask import Flask, request, jsonify
from google.cloud import pubsub_v1

app = Flask(__name__)
publisher = pubsub_v1.PublisherClient()
project_id = os.environ.get("PROJECT_ID")
topic_id = os.environ.get("TOPIC_ID")
topic_path = publisher.topic_path(project_id, topic_id)

@app.route("/orders", methods=["POST"])
def create_order():
    data = request.get_data()
    if not data:
        return jsonify({"error": "request body is required"}), 400
    future = publisher.publish(topic_path, data)
    message_id = future.result()
    return jsonify({"status": "queued", "message_id": message_id}), 202

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

data = request.get_data()

future = publisher.publish(topic_path, data)
message_id = future.result()

return jsonify({"status": "queued", "message_id": message_id}), 202

app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))

Flask==3.0.3
gunicorn==23.0.0
google-cloud-pubsub==2.28.0

cd order-api

gcloud run deploy order-api \
  --source . \
  --region $REGION \
  --service-account="$PRODUCER_SA@$PROJECT_ID.iam.gserviceaccount.com" \
  --set-env-vars PROJECT_ID=$PROJECT_ID,TOPIC_ID=$TOPIC_ID \
  --allow-unauthenticated

Building using Buildpacks and deploying container to Cloud Run service [order-api] in project [your-project-id] region [asia-northeast1]
✓ Building and deploying new service... Done.                                                                                      
  ✓ Validating configuration...                                                                                                    
  ✓ Uploading sources...                                                                                                           
  ✓ Building Container... Logs are available at [ https://console.cloud.google.com/cloud-build/builds;region=asia-northeast1/YOUR_BUILD_ID?project=YOUR_PROJECT_NUMBER ].                                                                           
  ✓ Creating Revision...                                                                                                           
  ✓ Routing traffic...                                                                                                             
  ✓ Setting IAM Policy...                                                                                                          
Done.                                                                                                                              
Service [order-api] revision [order-api-00001-4sl] has been deployed and is serving 100 percent of traffic.
Service URL: https://order-api-YOUR_PROJECT_NUMBER.asia-northeast1.run.app

# 注文リクエストの送信（レスポンスタイム計測付き）
curl -X POST $SERVICE_URL/orders \
  -H "Content-Type: application/json" \
  -d '{"order_id":"ORD-001","item":"book","qty":2}' \
  -w "\ntime_total: %{time_total}s\n"

curl -X POST $SERVICE_URL/orders \
  -H "Content-Type: application/json" \
  -d '{"order_id":"ORD-002","item":"pen","qty":5}' \
  -w "\ntime_total: %{time_total}s\n"

{"message_id":"19114283623792820","status":"queued"}

time_total: 0.208955s

{"message_id":"19114287379961629","status":"queued"}

time_total: 0.141060s