AWS 实现eventbridge 发送消息触发lamda

server/2024/10/23 9:33:33/

1 创建一个lamda函数，运行时选择python

贴入下面代码并点击deploy

import jsondef lambda_handler(event, context):# 打印接收到的事件print("Received event: " + json.dumps(event, indent=2))# 可以在这里添加其他处理逻辑# 例如，提取某些信息并进行处理# 返回响应return {'statusCode': 200,'body': json.dumps({'message': 'Event processed successfully'})}

2 创建事件总线

名字写上就可以保存了

在总线下进行规则创建

Event 有三种来源

AWS事件，合作伙伴事件，输入自己的

通过选择不同的事件来源，后面我们可以通过事件构造去匹配这种格式的event

比如选择自定义

在事件模式填入对应json然后点击测试，绿色说明测试成功可以匹配到该event

{"source": [{"wildcard": "*"}],"detail-type": [{"wildcard": "*"}],"detail": {"EventSource": [{"wildcard": "*"}],"eventSource": [{"wildcard": "*"}],"key1": [{"wildcard": "*"}]}
}

在本地准备python 代码, 其中EventBusName 指定event送到哪个总线里进行处理

import boto3
import json
from datetime import datetime# 创建 EventBridge 客户端
client = boto3.client('events')# 定义事件
event = {'Source': 'my.custom.source.test.bili.mysql',  # 事件来源'DetailType': 'AWS API Call via CloudTrail good or bad',  # 事件类型'Detail': json.dumps({"EventSource": "bili.custom.source","eventSource": "bili.custom.source","key1": "value1"}),'EventBusName': 'event-bus',  # 指定事件总线'Time': datetime.utcnow()  # 事件时间，使用 UTC
}# 发送事件
try:response = client.put_events(Entries=[event])print("Event sent! Response:", response)
except Exception as e:print("Error sending event:", e)

在lamda里添加监听器，选择evnetbridge-rule