如何使用Cloud Pubsub在文件上传至Google Cloud Compute Engine时发送邮件通知

通过在GCE实例上部署目录监控服务检测新文件上传，将文件信息推送到Pub/Sub主题，再通过Pub/Sub订阅触发Python脚本消费消息、发送邮件通知，全流程无外部依赖，可直接落地。

• 开启GCP项目的Pub/Sub API
• 创建GCP服务账号，绑定两个角色：roles/pubsub.publisher（GCE端发布消息用）、roles/pubsub.subscriber（Python脚本消费消息用）
• 下载服务账号JSON密钥文件，存到GCE实例的/opt/gcp-sa-key.json路径，执行chmod 600 /opt/gcp-sa-key.json限制权限
• 配置全局环境变量，将export GOOGLE_APPLICATION_CREDENTIALS="/opt/gcp-sa-key.json写入~/.bashrc后执行source ~/.bashrc生效

使用inotify-tools监控文件上传目录，检测到文件上传完成后自动推送消息到Pub/Sub

1. 安装依赖：

# Debian/Ubuntu系
sudo apt update && sudo apt install inotify-tools google-cloud-sdk -y

# CentOS/RHEL系
sudo yum install inotify-tools google-cloud-sdk -y

2. 新建监控脚本/opt/file-monitor.sh，替换内容里的GCP项目ID、监控目录为实际值：

#!/bin/bash
# 替换为你的文件上传目录
WATCH_DIR="/var/upload"
# 替换为你的GCP项目ID
GCP_PROJECT_ID="你的项目ID"
PUBSUB_TOPIC="projects/${GCP_PROJECT_ID}/topics/file-upload-notify-topic"

# 监控文件写入完成事件，过滤临时文件
inotifywait -m -e close_write --format '%w%f %T' --timefmt '%Y-%m-%d %H:%M:%S' "$WATCH_DIR" | while read FILE_PATH UPLOAD_TIME
do
    if [[ "$FILE_PATH" != *.tmp ]]; then
        FILE_NAME=$(basename "$FILE_PATH")
        MESSAGE=$(cat <<EOF
{
    "file_name": "$FILE_NAME",
    "file_path": "$FILE_PATH",
    "upload_time": "$UPLOAD_TIME",
    "server_hostname": "$(hostname)"
}
EOF
)
        # 推送消息到Pub/Sub
        gcloud pubsub topics publish "$PUBSUB_TOPIC" --message="$MESSAGE"
    fi
done

3. 给脚本加执行权限：chmod +x /opt/file-monitor.sh
4. 配置服务开机自启，新建/etc/systemd/system/file-monitor.service：

[Unit]
Description=File upload monitor service
After=network.target

[Service]
User=root
Environment="GOOGLE_APPLICATION_CREDENTIALS=/opt/gcp-sa-key.json"
ExecStart=/opt/file-monitor.sh
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

5. 启动服务：

sudo systemctl daemon-reload
sudo systemctl enable --now file-monitor.service

执行以下命令替换项目ID后运行：

# 创建主题
gcloud pubsub topics create file-upload-notify-topic --project=你的项目ID
# 创建订阅，消息确认超时设为30秒
gcloud pubsub subscriptions create file-upload-sub --topic=file-upload-notify-topic --ack-deadline=30 --project=你的项目ID

1. 安装Python依赖：

pip3 install google-cloud-pubsub python-dotenv

2. 新建配置文件/opt/.env，替换为实际的SMTP配置、GCP项目ID：

GCP_PROJECT_ID=你的GCP项目ID
PUBSUB_SUBSCRIPTION_ID=file-upload-sub
SMTP_SERVER=smtp.163.com
SMTP_PORT=465
SMTP_USER=你的发件人邮箱@163.com
SMTP_PASSWORD=你的SMTP授权码
RECIPIENT_EMAIL=收件人邮箱@xxx.com
GOOGLE_APPLICATION_CREDENTIALS=/opt/gcp-sa-key.json

3. 新建Python脚本/opt/pubsub-email-notify.py：

import os
import json
import smtplib
from email.mime.text import MIMEText
from google.cloud import pubsub_v1
from dotenv import load_dotenv

# 加载配置
load_dotenv()
project_id = os.getenv("GCP_PROJECT_ID")
subscription_id = os.getenv("PUBSUB_SUBSCRIPTION_ID")
smtp_server = os.getenv("SMTP_SERVER")
smtp_port = int(os.getenv("SMTP_PORT"))
smtp_user = os.getenv("SMTP_USER")
smtp_password = os.getenv("SMTP_PASSWORD")
recipient = os.getenv("RECIPIENT_EMAIL")

def send_email(file_info):
    subject = f"新文件上传通知：{file_info['file_name']}"
    content = f"""
    服务器主机名：{file_info['server_hostname']}
    文件完整路径：{file_info['file_path']}
    文件上传时间：{file_info['upload_time']}
    """
    msg = MIMEText(content, 'plain', 'utf-8')
    msg['Subject'] = subject
    msg['From'] = smtp_user
    msg['To'] = recipient

    try:
        with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
            server.login(smtp_user, smtp_password)
            server.sendmail(smtp_user, recipient, msg.as_string())
        return True
    except Exception as e:
        print(f"邮件发送失败：{str(e)}")
        return False

def message_callback(message: pubsub_v1.subscriber.message.Message) -> None:
    try:
        file_info = json.loads(message.data.decode('utf-8'))
        if send_email(file_info):
            # 发送成功确认消息
            message.ack()
        else:
            # 发送失败退回消息重试
            message.nack()
    except Exception as e:
        print(f"消息处理失败：{str(e)}")
        message.nack()

if __name__ == "__main__":
    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(project_id, subscription_id)
    stream_future = subscriber.subscribe(subscription_path, callback=message_callback)
    print(f"已启动订阅监听：{subscription_path}")

    try:
        stream_future.result()
    except KeyboardInterrupt:
        stream_future.cancel()
        stream_future.result()

4. 配置脚本服务开机自启，新建/etc/systemd/system/pubsub-email-notify.service：

[Unit]
Description=Pub/Sub email notify service
After=network.target

[Service]
User=root
WorkingDirectory=/opt
ExecStart=/usr/bin/python3 /opt/pubsub-email-notify.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

5. 启动服务：

sudo systemctl daemon-reload
sudo systemctl enable --now pubsub-email-notify.service

1. 上传任意文件到GCE的/var/upload目录，检查收件箱是否收到通知
2. 若未收到通知，执行以下命令查看日志排查：

# 查看目录监控服务日志
journalctl -u file-monitor.service -f
# 查看邮件通知服务日志
journalctl -u pubsub-email-notify.service -f

• 如需监控多级子目录，给inotifywait命令加上-r参数，同时执行echo "fs.inotify.max_user_watches=524288" >> /etc/sysctl.conf && sysctl -p调整系统最大监控数
• 如需避免重复发送邮件，可增加Redis去重逻辑：将已处理的文件路径+上传时间的哈希值存入Redis，设置24小时过期，处理消息前先校验是否已处理
• 可根据业务需求调整Pub/Sub的重试次数、死信队列配置，避免异常消息阻塞队列

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