Google PubSub问题：文件触发消息重复发布至两个订阅队列

核心问题 #

你的代码存在两个关键错误，导致所有消息会被两个订阅同时接收：

1. PubSub发布API参数错误：publisher.publish()方法并不支持subscription参数，你传入的这个参数不会被识别，所有消息都会直接发布到主题da-topic-dev-natdes。如果两个订阅没有配置过滤规则，自然会收到所有主题消息。
2. 缺少消息属性与订阅过滤配置：你没有给消息添加可用于路由的自定义属性，也没给订阅设置过滤规则，无法实现按文件夹分流的逻辑。

修复步骤 #

1. 修改Python发布代码 #

调整代码，给消息添加自定义属性（比如folder_type），用于后续订阅过滤：

# Publish messages to PubSub
publisher = pubsub.PublisherClient()
topic = publisher.topic_path('dev-data-31ce', 'da-topic-dev-natdes')
count = 0
signature = None
log.info('Identifying destination messages signature')

# 先判断文件所属文件夹，避免循环内重复判断
if 'ANALYTICS' in filename:
    signature = 'Analytics'
else:
    signature = 'Marketing'

for line in lines:
    data = line.decode()
    m_data = json.dumps(data).encode('utf-8')
    
    # 给消息添加自定义属性，用于订阅过滤
    future = publisher.publish(
        topic,
        data=m_data,
        folder_type=signature  # 自定义属性，值为文件夹类型
    )
    count += 1

优化点：把文件夹判断移到循环外，避免重复执行相同判断，提升效率

2. 配置PubSub订阅过滤规则 #

在Google Cloud控制台或通过gcloud命令，给两个订阅分别设置过滤规则：

• Analytics订阅：设置过滤规则为 attributes.folder_type = "Analytics"，只接收标记为Analytics的消息
• Marketing订阅：设置过滤规则为 attributes.folder_type = "Marketing"，只接收标记为Marketing的消息

3. 额外检查项 #

• 确认filename变量确实包含文件夹路径信息，比如是否为ANALYTICS/xxx.csv这类格式，避免因路径判断不准确导致路由错误
• 如果文件夹名称存在大小写差异，可将判断逻辑改为if 'ANALYTICS' in filename.upper()，增强兼容性

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