Google Pubsub Python客户端库订阅者随机崩溃求助

Hey there, let's dig into this Pub/Sub issue you're facing—silent errors are always the trickiest! Since you mentioned following the async pull tutorial but hitting a wall without any feedback, let's break down common culprits and actionable steps to debug this:

1. Add Explicit Logging & Exception Handling #

Silent failures often happen because exceptions aren't being caught or logged, especially with async code where unhandled coroutine errors can get swallowed. Add this structure to your subscriber code to force visibility into what's going wrong:

import logging
import asyncio
from google.cloud import pubsub_v1

# Set up detailed logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def callback(message):
    try:
        logger.info(f"Received message: {message.data.decode('utf-8')}")
        # Your message processing logic here
        message.ack()
    except Exception as e:
        logger.error(f"Failed to process message: {str(e)}", exc_info=True)
        # Let Pub/Sub retry the message instead of acknowledging it
        message.nack()

async def main():
    subscriber = pubsub_v1.AsyncSubscriberClient()
    subscription_path = subscriber.subscription_path("YOUR_PROJECT_ID", "YOUR_SUBSCRIPTION_ID")
    
    try:
        streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
        logger.info(f"Listening for messages on {subscription_path}...")
        # Keep the listener running indefinitely
        await streaming_pull_future.result()
    except Exception as e:
        logger.error(f"Subscriber connection failed: {str(e)}", exc_info=True)
        streaming_pull_future.cancel()
        await streaming_pull_future.cancelled()

if __name__ == "__main__":
    asyncio.run(main())

This will log every step, including connection issues, permission errors, or processing failures that might be happening behind the scenes.

2. Verify GCE Instance Permissions #

Since your subscriber runs on a Compute Engine instance, the instance's service account needs the right Pub/Sub permissions:

• Ensure the service account has the roles/pubsub.subscriber role assigned to your subscription (use roles/pubsub.editor temporarily for testing if needed).
• Check if the Pub/Sub API is enabled for your project (you can confirm this in the Cloud Console's API library).
• Test permissions directly from the GCE instance with this command:

gcloud projects get-iam-policy YOUR_PROJECT_ID --filter="bindings.members:serviceAccount:INSTANCE_SERVICE_ACCOUNT@YOUR_PROJECT_ID.iam.gserviceaccount.com"

Replace INSTANCE_SERVICE_ACCOUNT with the service account attached to your GCE instance (the default compute engine service account is used if you didn't specify a custom one).

3. Validate Topic & Subscription Alignment #

• Double-check that your sendmessage.py is publishing to the exact same topic your subscription is linked to—even a tiny typo here means your subscriber will never see messages.
• Confirm your subscription is set to pull mode (not push) since you're using the pull client library.
• Check the Cloud Console's Pub/Sub metrics for your topic/subscription: if messages are queued up but not being pulled, that points to a subscriber-side issue.

4. Confirm Async Client Setup #

• Make sure you're using pubsub_v1.AsyncSubscriberClient (not the synchronous SubscriberClient)—mixing these up can cause unexpected silent failures.
• If you're using a custom service account key (instead of the GCE instance's default), ensure the key path is correctly set:

  export GOOGLE_APPLICATION_CREDENTIALS="/path/to/your/service-account-key.json"
  

  Or set it directly in code:

  subscriber = pubsub_v1.AsyncSubscriberClient.from_service_account_json("/path/to/key.json")
  

Once you add logging and work through these checks, you should start seeing clear error messages that point to the root cause—whether it's a permission gap, misconfigured subscription, or unhandled exception in your callback.

内容的提问来源于stack exchange，提问作者Paul