AWS Lambda CloudWatch Jobs TypeScript

需要自动化定期任务如备份、邮件发送或数据处理？AWS Lambda结合CloudWatch Events/EventBridge是替代传统cron作业的无服务器完美解决方案。本指南专注于TypeScript/Node.js的生产级定时任务实现。

Lambda定时任务概述#

AWS Lambda是无服务器计算服务，允许您在不管理服务器的情况下运行代码。当与调度服务结合时，您可以：

自动化任务：备份、报告生成、数据清理
节省成本：仅在代码执行时付费
自动扩展：无需配置即可处理负载增加
高可用性：AWS确保可用性

[!NOTE] AWS目前推荐使用EventBridge Scheduler而不是传统的CloudWatch Events用于新的定时任务。

flowchart LR
    subgraph Scheduler["调度触发器"]
        direction TB
        EBS["EventBridge Scheduler<br/>(时区支持 / 重试 / 灵活时间窗口)"]
        CWE["CloudWatch Events<br/>(传统Cron)"]
    end

    subgraph Serverless["无服务器计算"]
        Lambda["AWS Lambda Function<br/>(Node.js / TypeScript)"]
        DLQ["Dead Letter Queue (SQS)<br/>(调用失败死信队列)"]
    end

    subgraph Targets["目标资源与监控"]
        DB[(数据库 / S3存储桶)]
        Log["CloudWatch Logs / 告警"]
    end

    EBS -->|定时触发| Lambda
    CWE -.->|基础触发| Lambda
    Lambda -.->|执行失败 / 超过重试上限| DLQ
    Lambda -->|读写数据| DB
    Lambda -->|记录日志与指标| Log

EventBridge Scheduler是现代调度服务，与CloudWatch Events相比具有增强功能。

EventBridge Scheduler的优势#

时区支持：不仅限于UTC
灵活时间窗口：在灵活时间范围内运行
内置重试：失败时自动重试
一次性调度：支持单次执行
死信队列：更好的错误处理

步骤1: 创建TypeScript Lambda函数#

lambda-handler.ts

import { Context } from 'aws-lambda';
import { DynamoDB } from 'aws-sdk';
import { logger } from './utils/logger';

interface ScheduledEvent {
  time: string;
  detail?: any;
}

interface LambdaResponse {
  statusCode: number;
  body: string;
}

export const lambdaHandler = async (
  event: ScheduledEvent,
  context: Context
): Promise<LambdaResponse> => {
  logger.info(`任务触发时间: ${new Date().toISOString()}`);
  logger.info(`事件: ${JSON.stringify(event)}`);
  
  try {
    // 您的任务处理逻辑
    const result = await processScheduledTask(event);
    
    return {
      statusCode: 200,
      body: JSON.stringify({
        message: '任务成功完成',
        result
      })
    };
  } catch (error) {
    logger.error(`任务失败: ${error}`);
    throw error;
  }
};

async function processScheduledTask(event: ScheduledEvent): Promise<any> {
  // 示例：清理数据库、发送报告等
  logger.info('处理定时任务...');
  
  // 您的业务逻辑在这里
  return { status: 'success', processedItems: 10 };
}

typescript

步骤2: 创建IAM角色#

Lambda需要权限来执行任务和写入日志：

trust-policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

json

策略权限：

permissions-policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:*",
        "s3:*"
      ],
      "Resource": "*"
    }
  ]
}

json

步骤3: 使用EventBridge Scheduler创建调度#

使用AWS控制台#

进入EventBridge Scheduler控制台
点击创建调度
选择定期调度
配置调度表达式：

# 每天上午9:00 UTC运行
cron(0 9 * * ? *)

# 每15分钟运行
rate(15 minutes)

# 每周一上午9:00运行
cron(0 9 ? * MON *)

# 每月第一天运行
cron(0 0 1 * ? *)

bash

选择Lambda函数作为目标
配置时区（如果需要）：

# 东部时间上午9:00
cron(0 9 ? * MON *)
timezone: America/New_York

bash

配置灵活时间窗口（可选）：

# 在调度时间后15分钟内运行
mode: FLEXIBLE
maximum_window_in_minutes: 15

bash

设置重试策略：

maximum_retry_attempts: 3
maximum_event_age_in_seconds: 3600

bash

使用AWS CLI#

aws scheduler create-schedule \
  --name daily-job \
  --schedule-expression 'cron(0 9 * * ? *)' \
  --schedule-expression-timezone 'UTC' \
  --target '{
    "Arn": "arn:aws:lambda:us-east-1:123456789012:function:my-function",
    "RoleArn": "arn:aws:iam::123456789012:role/scheduler-role"
  }' \
  --flexible-time-window '{
    "Mode": "FLEXIBLE",
    "MaximumWindowInMinutes": 15
  }'

bash

使用Terraform#

scheduler.tf

resource "aws_scheduler_schedule" "daily_job" {
  name       = "daily-job"
  group_name = "default"

  flexible_time_window {
    mode                      = "FLEXIBLE"
    maximum_window_in_minutes = 15
  }

  schedule_expression          = "cron(0 9 * * ? *)"
  schedule_expression_timezone = "UTC"

  target {
    arn      = aws_lambda_function.my_function.arn
    role_arn = aws_iam_role.scheduler_role.arn

    retry_policy {
      maximum_retry_attempts       = 3
      maximum_event_age_in_seconds = 3600
    }

    dead_letter_config {
      arn = aws_sqs_queue.dlq.arn
    }
  }
}

hcl

方法2: CloudWatch Events（传统）#

CloudWatch Events是传统方法，仍然支持但功能较少。

步骤1: 创建Lambda函数#

与Scheduler方法相同。

步骤2: 创建CloudWatch规则#

使用AWS控制台#

进入CloudWatch控制台
选择Events → Rules
点击创建规则
选择调度表达式
输入cron表达式：

# 每天上午9:00 UTC
0 9 * * ? *

# 每5分钟
rate(5 minutes)

bash

选择Lambda函数作为目标
配置权限（AWS将自动创建）

使用AWS CLI#

aws events put-rule \
  --name daily-lambda-rule \
  --schedule-expression 'cron(0 9 * * ? *)'

aws events put-targets \
  --rule daily-lambda-rule \
  --targets '{
    "Id": "1",
    "Arn": "arn:aws:lambda:us-east-1:123456789012:function:my-function"
  }'

aws lambda add-permission \
  --function-name my-function \
  --statement-id daily-lambda-rule \
  --action 'lambda:InvokeFunction' \
  --principal events.amazonaws.com \
  --source-arn arn:aws:events:us-east-1:123456789012:rule/daily-lambda-rule

bash

Cron表达式指南#

EventBridge使用6字段cron表达式：

cron(分钟 小时 日-月 月 星期-日 年)

text

Cron表达式示例#

# 每天上午9:00 UTC
cron(0 9 * * ? *)

# 每周一上午9:00
cron(0 9 ? * MON *)

# 每月1日00:00
cron(0 0 1 * ? *)

# 每15分钟
rate(15 minutes)

# 每1小时
rate(1 hour)

# 周一到周五上午9:00
cron(0 9 ? * MON-FRI *)

# 月的最后一天
cron(0 0 L * ? *)

# 每月的第一个星期一
cron(0 0 ? * MON#1 *)

bash

字段值#

字段	值	特殊字符
分钟	0-59	, - * /
小时	0-23	, - * /
日-月	1-31, L, W	, - * / ? L W
月	1-12, JAN-DEC	, - * /
星期-日	1-7, SUN-SAT, ?, L, #	, - * / ? L #
年	1970-2199	, - * /

[!WARNING] 当日-月或星期-日受约束时使用?，不要同时使用两个字段。

最佳实践#

1. 幂等性#

确保处理程序是幂等的，以便安全重试：

idempotent-handler.ts

import { DynamoDB } from 'aws-sdk';

const dynamodb = new DynamoDB.DocumentClient();
const TABLE_NAME = 'job-locks';

export const lambdaHandler = async (event: ScheduledEvent): Promise<any> => {
  const jobId = `${event.time}-${event.detail?.id || 'default'}`;
  
  // 检查任务是否已运行
  if (await isJobAlreadyProcessed(jobId)) {
    logger.info('任务已处理，跳过');
    return { status: 'skipped' };
  }
  
  // 处理任务
  const result = await processJob(event);
  
  // 标记任务为已处理
  await markJobAsProcessed(jobId);
  
  return result;
};

async function isJobAlreadyProcessed(jobId: string): Promise<boolean> {
  try {
    const result = await dynamodb.get({
      TableName: TABLE_NAME,
      Key: { jobId }
    }).promise();
    return !!result.Item;
  } catch (error) {
    logger.error(`检查任务状态错误: ${error}`);
    return false;
  }
}

async function markJobAsProcessed(jobId: string): Promise<void> {
  await dynamodb.put({
    TableName: TABLE_NAME,
    Item: {
      jobId,
      processedAt: new Date().toISOString(),
      ttl: Math.floor(Date.now() / 1000) + (7 * 24 * 60 * 60) // 7天TTL
    }
  }).promise();
}

typescript

2. 死信队列（DLQ）#

设置DLQ来处理失败的调用：

dlq-handler.ts

import { SQS } from 'aws-sdk';

const sqs = new SQS();
const DLQ_URL = process.env.DLQ_URL || '';

async function sendToDLQ(errorMessage: string, event: ScheduledEvent): Promise<void> {
  await sqs.sendMessage({
    QueueUrl: DLQ_URL,
    MessageBody: JSON.stringify({
      error: errorMessage,
      event,
      timestamp: new Date().toISOString()
    })
  }).promise();
}

export const lambdaHandler = async (event: ScheduledEvent): Promise<any> => {
  try {
    const result = await processJob(event);
    return result;
  } catch (error) {
    const errorMessage = error instanceof Error ? error.message : String(error);
    logger.error(`任务失败: ${errorMessage}`);
    
    // 发送到DLQ
    await sendToDLQ(errorMessage, event);
    
    throw error;
  }
};

typescript

3. 防止重叠执行#

使用DynamoDB锁防止重叠执行：

lock-mechanism.ts

import { DynamoDB } from 'aws-sdk';

const dynamodb = new DynamoDB.DocumentClient();
const LOCK_TABLE = 'job-locks';

interface LockItem {
  jobId: string;
  lockedAt: string;
  expiresAt: string;
}

async function acquireLock(jobId: string): Promise<boolean> {
  const now = new Date();
  const expiresAt = new Date(now.getTime() + 10 * 60 * 1000); // 10分钟
  
  try {
    await dynamodb.put({
      TableName: LOCK_TABLE,
      Item: {
        jobId,
        lockedAt: now.toISOString(),
        expiresAt: expiresAt.toISOString()
      },
      ConditionExpression: 'attribute_not_exists(jobId)'
    }).promise();
    
    return true;
  } catch (error) {
    if ((error as any).code === 'ConditionalCheckFailedException') {
      return false; // 锁已存在
    }
    throw error;
  }
}

async function releaseLock(jobId: string): Promise<void> {
  await dynamodb.delete({
    TableName: LOCK_TABLE,
    Key: { jobId }
  }).promise();
}

export const lambdaHandler = async (event: ScheduledEvent): Promise<any> => {
  const jobId = `scheduled-job-${event.time}`;
  
  // 获取锁
  const lockAcquired = await acquireLock(jobId);
  if (!lockAcquired) {
    logger.info('任务正在运行，跳过');
    return { status: 'skipped', reason: 'lock_not_acquired' };
  }
  
  try {
    const result = await processJob(event);
    return result;
  } finally {
    // 释放锁
    await releaseLock(jobId);
  }
};

typescript

4. 监控和日志#

使用CloudWatch Logs和Metrics：

monitoring-handler.ts

import { CloudWatch } from 'aws-sdk';

const cloudwatch = new CloudWatch({ region: process.env.AWS_REGION });

export const lambdaHandler = async (event: ScheduledEvent, context: Context): Promise<any> => {
  const startTime = Date.now();
  
  try {
    // 自定义指标
    logger.info('JOB_STARTED');
    
    const result = await processJob(event);
    
    // 记录持续时间
    const duration = Date.now() - startTime;
    logger.info(`JOB_COMPLETED duration=${duration}ms`);
    
    // 发送自定义指标
    await cloudwatch.putMetricData({
      Namespace: 'ScheduledJobs',
      MetricData: [
        {
          MetricName: 'JobDuration',
          Value: duration,
          Unit: 'Milliseconds',
          Dimensions: [
            {
              Name: 'JobName',
              Value: 'DailyCleanup'
            }
          ]
        }
      ]
    }).promise();
    
    return result;
  } catch (error) {
    const duration = Date.now() - startTime;
    logger.error(`JOB_FAILED error=${error} duration=${duration}ms`);
    
    // 发送失败指标
    await cloudwatch.putMetricData({
      Namespace: 'ScheduledJobs',
      MetricData: [
        {
          MetricName: 'JobFailures',
          Value: 1,
          Unit: 'Count',
          Dimensions: [
            {
              Name: 'JobName',
              Value: 'DailyCleanup'
            }
          ]
        }
      ]
    }).promise();
    
    throw error;
  }
};

typescript

5. 错误处理#

优雅地处理错误：

error-handling.ts

interface APIResponse {
  statusCode: number;
  body: string;
}

export const lambdaHandler = async (event: ScheduledEvent): Promise<APIResponse> => {
  try {
    const result = await processJob(event);
    return {
      statusCode: 200,
      body: JSON.stringify(result)
    };
  } catch (error) {
    if (error instanceof ValidationError) {
      logger.error(`验证错误: ${error.message}`);
      return {
        statusCode: 400,
        body: JSON.stringify({ error: error.message })
      };
    } else if (error instanceof BusinessError) {
      logger.error(`业务错误: ${error.message}`);
      return {
        statusCode: 422,
        body: JSON.stringify({ error: error.message })
      };
    } else {
      logger.error(`意外错误: ${error}`);
      // 发送到DLQ或警报
      await sendAlert(error instanceof Error ? error.message : String(error));
      throw error;
    }
  }
};

class ValidationError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'ValidationError';
  }
}

class BusinessError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'BusinessError';
  }
}

typescript

比较: EventBridge Scheduler vs CloudWatch Events#

功能	EventBridge Scheduler	CloudWatch Events
调度类型	cron + rate + one-time	cron + rate
时区支持	✅ 任何时区	❌ 仅UTC
灵活窗口	✅ 是	❌ 否
内置重试	✅ 是	❌ 否（需要DLQ）
成本	$1.00/M调用	免费（前5M/月）
用例	生产任务	简单定期任务

实践示例: 每日数据库清理#

database-cleanup.ts

import { DynamoDB } from 'aws-sdk';

const dynamodb = new DynamoDB.DocumentClient();
const TABLE_NAME = 'user-activity';

interface CleanupResult {
  deletedCount: number;
  cutoffDate: string;
}

export const lambdaHandler = async (): Promise<APIResponse> => {
  const cutoffDate = new Date();
  cutoffDate.setDate(cutoffDate.getDate() - 30);
  
  try {
    // 扫描并删除旧记录
    const response = await dynamodb.scan({
      TableName: TABLE_NAME,
      FilterExpression: 'created_at < :cutoff',
      ExpressionAttributeValues: {
        ':cutoff': cutoffDate.toISOString()
      }
    }).promise();
    
    let deletedCount = 0;
    
    if (response.Items) {
      for (const item of response.Items) {
        await dynamodb.delete({
          TableName: TABLE_NAME,
          Key: { id: item.id }
        }).promise();
        deletedCount++;
      }
    }
    
    logger.info(`删除了${deletedCount}条旧记录`);
    
    return {
      statusCode: 200,
      body: JSON.stringify({
        deletedCount,
        cutoffDate: cutoffDate.toISOString()
      } as CleanupResult)
    };
  } catch (error) {
    logger.error(`清理失败: ${error}`);
    throw error;
  }
};

typescript

使用EventBridge Scheduler调度：

# 每天凌晨2:00 UTC运行
cron(0 2 * * ? *)

bash

故障排除#

任务不运行#

检查调度表达式：确保cron语法正确
验证IAM权限：Lambda具有必要权限
检查CloudWatch Logs：查找错误消息
验证目标ARN：确保Lambda函数ARN正确

任务运行但失败#

查看CloudWatch Logs：查找错误消息
检查超时：Lambda可能超时
验证资源权限：数据库、S3等访问权限
本地测试：本地运行Lambda进行调试

重叠执行#

实现锁机制：使用DynamoDB或Redis
使用预留并发：限制并发执行
添加幂等性：确保处理程序是幂等的

清理资源#

不再需要时，清理资源：

# 删除调度
aws scheduler delete-schedule --name daily-job

# 删除CloudWatch规则
aws events delete-rule --name daily-lambda-rule

# 删除Lambda函数
aws lambda delete-function --function-name my-function

# 删除IAM角色
aws iam delete-role --role-name scheduler-role

bash

结论#

AWS Lambda结合EventBridge Scheduler/CloudWatch Events为定时任务提供了强大的解决方案：

EventBridge Scheduler：用于具有时区支持、重试、灵活窗口的生产任务
CloudWatch Events：用于具有成本效益的简单定期任务

关键要点：

生产环境使用EventBridge Scheduler
实现幂等性和错误处理
使用CloudWatch设置监控
为失败的调用使用DLQ
防止重叠执行

Lambda定时任务概述#

EventBridge Scheduler的优势#

步骤1: 创建TypeScript Lambda函数#

步骤2: 创建IAM角色#

步骤3: 使用EventBridge Scheduler创建调度#

使用AWS控制台#

使用AWS CLI#

使用Terraform#

方法2: CloudWatch Events（传统）#

步骤1: 创建Lambda函数#

步骤2: 创建CloudWatch规则#

使用AWS控制台#

使用AWS CLI#

Cron表达式指南#

Cron表达式示例#

字段值#

最佳实践#

1. 幂等性#

2. 死信队列（DLQ）#

3. 防止重叠执行#

4. 监控和日志#

5. 错误处理#

比较: EventBridge Scheduler vs CloudWatch Events#

实践示例: 每日数据库清理#

故障排除#

任务不运行#

任务运行但失败#

重叠执行#

清理资源#

结论#

参考文献#