如何通过定时器在Google Cloud Functions中定时运行Python 3.x代码

我来分享两种GCP原生支持的方案，帮你实现Python代码每小时自动运行，配置起来都挺直观的：

方案一：Cloud Scheduler 直接触发HTTP类型Cloud Function #

这是最直接的方式，适合简单的定时任务场景：

1. 编写并部署HTTP触发的Python函数
   先写一个HTTP类型的函数，示例代码如下：

   import functions_framework
   
   @functions_framework.http
   def hourly_scheduled_task(request):
       # 这里替换成你的每小时任务逻辑
       print("开始执行每小时定时任务...")
       # 比如数据同步、报表生成等操作
       return "定时任务执行完成", 200
   

   部署时选择HTTP触发，注意生产环境不要开启“允许未认证调用”，后续通过服务账号授权Scheduler调用。

2. 创建Cloud Scheduler定时任务

   • 打开Cloud Scheduler控制台，点击「创建作业」
   • 填写作业名称（比如hourly-python-function-job），频率填Cron表达式0 * * * *（代表每小时整点执行）
   • 目标类型选「HTTP」，填入你的Cloud Function触发URL（可在Cloud Functions控制台的函数详情页找到）
   • HTTP方法选「GET」（如果你的函数需要处理POST，也可以选POST并添加请求体）
   • 身份验证选「服务账号」，指定一个拥有roles/cloudfunctions.invoker权限的服务账号，确保Scheduler能合法调用函数
   • 保存后，任务就会按设定的频率自动触发你的Python函数了

方案二：Cloud Scheduler + Cloud Pub/Sub 触发Cloud Function #

这个方案适合需要解耦的场景，或者你的函数原本就是Pub/Sub触发类型：

1. 创建Pub/Sub主题
   在Cloud Pub/Sub控制台创建一个主题，比如hourly-task-topic，用于传递定时触发信号。

2. 编写并部署Pub/Sub触发的Python函数
   写一个监听Pub/Sub消息的函数，示例代码：

   import base64
   
   def hourly_pubsub_task(event, context):
       """处理来自Pub/Sub的定时触发消息"""
       if event.get('data'):
           msg_content = base64.b64decode(event['data']).decode('utf-8')
           print(f"收到触发消息：{msg_content}")
       # 执行你的定时任务逻辑
       print("执行每小时定时任务...")
   

   部署时选择Pub/Sub触发，关联刚才创建的主题。

3. 创建Cloud Scheduler定时任务

   • 同样在Cloud Scheduler控制台创建作业，频率填0 * * * *
   • 目标类型选「Pub/Sub」，选择之前创建的主题，消息内容可以自定义（比如"Trigger hourly task"）
   • 确保Scheduler使用的服务账号拥有roles/pubsub.publisher权限，这样它才能向主题发送消息
   • 保存后，每小时Scheduler会向Pub/Sub主题发消息，自动触发你的函数

一些关键注意事项 #

• 权限配置：不管用哪种方案，一定要给Scheduler的服务账号配置对应权限，否则会触发失败
• 超时调整：如果你的任务执行时间较长，记得在Cloud Functions控制台调整函数的超时时间（默认90秒，最长可设为540秒）
• 日志排查：如果任务没按预期执行，去Cloud Logging查看函数或Scheduler的日志，很容易定位问题
• Cron表达式验证：不确定表达式是否正确？创建Scheduler作业时，控制台会实时验证并显示下几次执行时间

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