Google Cloud Function中解码Pub/Sub消息时Base64解码失效问题

本地Python代码将消息做Base64编码后发布到Google Pub/Sub，本地解码能正常得到原始内容，但订阅该Topic的Cloud Function执行解码后，输出仍是编码后的字符串，无法还原原始内容。

本地编码发布及解码代码 #

pubMessage = ptp_jsonObject['restaurant']
message_bytes = str(pubMessage).encode('utf-8')
bas64_bytes = base64.b64encode(message_bytes)
print("PRINTING THE ENCODED MESSAGE")
print(bas64_bytes)
# OUTPUT: b'eydSJzogeydyZXNfaWQnOiAnMTY1MjA0MjYnfSwgJ2lkJzogJzE2NTIwNDI2JywgJ25hbWUnOiAnQnJvdGhlciBIdWJiYXJkJywgJ2xvY2FsaXR5X3ZlcmJvc2UnOiAnTm9ydGggQ2l0eSwgRHVibGluJ30='

publish_future = ptp_publisher_client.publish(ptp_topic_path, data=bas64_bytes)
result = publish_future.result()
print('Successfully published the event to pubsub')

base64_message = base64.b64decode(bas64_bytes).decode('utf-8')
print("PRINTING THE DECODED MESSAGE")
print(base64_message)
# OUTPUT: {'R': {'res_id': '16520426'}, 'id': '16520426', 'name': 'Brother Hubbard', 'locality_verbose': 'North City, Dublin'}

Pub/Topic中显示的消息内容 #

eydSJzogeydyZXNfaWQnOiAnMTY1MjA0MjYnfSwgJ2lkJzogJzE2NTIwNDI2JywgJ25hbWUnOiAnQnJvdGhlciBIdWJiYXJkJywgJ2xvY2FsaXR5X3ZlcmJvc2UnOiAnTm9ydGggQ2l0eSwgRHVibGluJ30=

Cloud Function订阅解码代码 #

import base64
import json

def hello_pubsub(event, context):
    pmessage = event['data']
    pubsub_message = base64.b64decode(pmessage).decode('utf-8')
    print('PRINTING DECODED MESSAGE PUBSUB')
    print(pubsub_message)
# OUTPUT: eydSJzogeydyZXNfaWQnOiAnMTY1MjA0MjYnfSwgJ2lkJzogJzE2NTIwNDI2JywgJ25hbWUnOiAnQnJvdGhlciBIdWJiYXJkJywgJ2xvY2FsaXR5X3ZlcmJvc2UnOiAnTm9ydGggQ2l0eSwgRHVibGluJ30=

问题原因 #

Google Pub/Sub的Python客户端publish方法在处理data参数时，如果传入的是bytes类型，会自动对其进行Base64编码，用于消息的传输存储。而你在本地代码中已经手动对消息做了一次Base64编码，导致最终发布到Pub/Sub的消息被两次Base64编码。

Cloud Function中订阅到的消息是经过客户端自动编码后的结果，此时你只解码一次，得到的是你手动编码后的字符串，而非原始内容。

解决方案 #

有两种处理方式，任选其一即可：

方式一：移除本地手动编码，让Pub/Sub客户端自动处理

修改本地发布代码，直接传入原始字符串的bytes，不需要手动做Base64编码：

pubMessage = ptp_jsonObject['restaurant']
# 直接将字符串转为bytes，不手动编码
message_bytes = str(pubMessage).encode('utf-8')
# 直接传入message_bytes，客户端会自动做Base64编码
publish_future = ptp_publisher_client.publish(ptp_topic_path, data=message_bytes)
result = publish_future.result()
print('Successfully published the event to pubsub')

此时Cloud Function中的原有解码代码可以正常工作，一次解码就能得到原始内容。

方式二：保留本地手动编码，Cloud Function中解码两次

如果必须保留本地的手动Base64编码操作，那么Cloud Function中需要执行两次Base64解码：

import base64
import json

def hello_pubsub(event, context):
    pmessage = event['data']
    # 第一次解码得到手动编码后的bytes，第二次解码得到原始内容
    first_decode = base64.b64decode(pmessage)
    pubsub_message = base64.b64decode(first_decode).decode('utf-8')
    print('PRINTING DECODED MESSAGE PUBSUB')
    print(pubsub_message)

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