Python通过代理拉取GCP PubSub消息时遇max_messages参数无效错误

报错信息 #

json returned "You have passed an invalid argument to the service (argument=max_messages)." Details: "You have passed an invalid argument to the service (argument=max_messages)."

问题原因 #

调用Pub/Sub的pull接口时，必须指定max_messages参数（对应API规范中的maxMessages），该参数用于定义单次拉取的最大消息数量，未传入会触发参数无效错误。

解决方案 #

修改pull_messages函数中的请求代码，添加max_messages参数，同时可根据需求设置return_immediately参数（控制是否立即返回，即使没有可用消息）。另外原代码中消息数据为base64编码，建议解码后再处理，避免乱码；处理完消息后记得确认，防止重复拉取。

修改后的完整代码：

import json
from googleapiclient.discovery import build
from httplib2 import Http
import httplib2
from oauth2client.service_account import ServiceAccountCredentials
import base64

# Replace placeholders with your project ID, topic name, subscription name, and proxy details
project_id = 'v-acp'  # Replace with your GCP project ID
topic_name = 'd_ack'   # Replace with your Pub/Sub topic name
subscription_name = 'dpull'  # Replace with your desired subscription name
proxy_host = '192.173.10.2'  # Replace with your proxy server address
proxy_port = 8095  # Replace with your proxy server port

# Create credentials (replace with your own authentication method)
credentials = ServiceAccountCredentials.from_json_keyfile_name(
    'key.json',
    scopes=['https://www.googleapis.com/auth/pubsub']
)

# Configure HTTP connection with proxy
proxy_info = httplib2.ProxyInfo(proxy_type=httplib2.socks.PROXY_TYPE_HTTP_NO_TUNNEL,
                                proxy_host=proxy_host,
                                proxy_port=proxy_port)
http = Http(proxy_info=proxy_info)

# Build the Pub/Sub API client
service = build('pubsub', 'v1', http=credentials.authorize(http))

# Pull messages from the subscription
def pull_messages():
    request = service.projects().subscriptions().pull(
        subscription=f'projects/{project_id}/subscriptions/{subscription_name}',
        body={
            'maxMessages': 10,  # 单次拉取最大消息数，取值范围1-1000，可按需调整
            'returnImmediately': True  # 无消息时立即返回，设为false则等待有消息再返回
        }
    )
    response = request.execute()
    if 'receivedMessages' in response:
        for message in response['receivedMessages']:
            # 解码base64格式的消息数据
            decoded_data = base64.b64decode(message['message']['data']).decode('utf-8')
            print(f"Received message: {decoded_data}")
            # 确认消息，避免重复拉取
            ack_request = service.projects().subscriptions().acknowledge(
                subscription=f'projects/{project_id}/subscriptions/{subscription_name}',
                body={'ackIds': [message['ackId']]}
            )
            ack_request.execute()

# Call the pull_messages function to retrieve messages
pull_messages()

关键说明 #

• max_messages取值范围为1到1000，可根据业务场景调整。
• returnImmediately设为true时，无可用消息会直接返回空结果；设为false时，会等待到有消息或超时才返回。
• 消息确认步骤不可省略，否则消息会在订阅的可见性超时后重新被拉取。

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