系统中间件与云虚拟化-serverless-基于阿里云函数计算的云工作流CloudFlow设计与体验

JAY.LIN 收录于系统中间件与云虚拟化

2025-09-16 约 1945 字预计阅读 4 分钟

https://bing.ee123.net/img/rand?artid=151760549

系统中间件与云虚拟化-serverless-基于阿里云函数计算的云工作流CloudFlow设计与体验

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

前言

1. 实验系列

·Serverless & DevOps 系列

2. 实验目的

·理解并掌握阿里云CloudFlow的基本概念，及其与函数计算的可视化编排工具，能够按需设计CloudFlow，通
过编排多个函数计算的自动执行工作流，从而构建服务接口

3. 实验要求

本实验将设计并实现一个基于阿里云CloudFlow和FC的温湿度数据监控及告警模拟系统。系统要求在温湿度
数据上报超过预设阀值时，通过调用告警接口发送邮件通知管理员。具体任务如下：
1，构建并部署数据上报接口：实现一个新的FC，作为数据上报接口，通过HTTP接收模拟的温湿度数据。在该函
数中，对温湿度值进行判断：
。如果未超过预设的安全阈值，返回正常状态。
。如果超过预设阈值，则调用实验一中的告警FC，向管理员发送告警邮件。
2.编排云工作流：创建阿里云CloudFlow，并将数据上报FC和实验一中的告警FC进行编排，使它们按顺序执
行，实现自动化的数据监控和告警流程。
3．云工作流测试：
·创建一个HTTP形式的工作流调度，作为对外暴露数据上报入口。
。使用Apifox工具模拟边缘设备向云端上报温湿度数据，通过CloudFlow平台的可视化界面查看工作流的执
行情况，体验CloudFlow在编排FC时的可视化优势。

4. 实验环境

4.1 获取云工作流免费额度

4.2 创建用于执行云工作流的角色

点击创建角色

因为是给云工作流提供的角色，所以选云服务

一个是 AliyunFCFullAccess，用于管理函数计算，调用FC

AliyunFnFFullAccess
用于管理函数工作流，用于操作工作流
AliyunEventBridgeFullAccess管理事件总线，用于后续创建工作流调度

4.3 开通事件总线EventBridge

免费开通
然后对EventBridge授权，才可以接收事件

5. 核心步骤

5.1 构建并部署数据上传接口

import datetime

from sanic import Sanic
from sanic.response import json

app = Sanic("MyApp")

#温度阈值
t_threshold = (25,28)
#湿度阈值
h_threshold = (30,33)

# if t_out_flag or h_out_flag:
#     "status": 1
# else:
#     "status": 0

@app.route("/upload",methods=["POST"])
async def data_upload(request):
    try:
        data = request.json
        sn = data.get("sn")
        temperature = data.get("temperature")
        humidity = data.get("humidity")
        if not all([sn,temperature,humidity]):
            return json({"error": "Missing sn, temperature, or humidity"}, status=400)
        #判断温湿度是否超过阈值
        #t_out_flag为true就表示超过了阈值
        t_out_flag = not (
            t_threshold[0]<=temperature<=t_threshold[1])
        h_out_flag = not (
            h_threshold[0]<=humidity<=h_threshold[1])
        email_body = generate_email_body(
            sn,
            temperature,
            humidity,t_threshold,h_threshold
        )
        res = {
            "status" : 1 if t_out_flag or h_out_flag else 0,
            "message" : "异常" if t_out_flag or h_out_flag else "正常",
            "data" : {
                #需要自定义收件人邮箱
                "recipient":"zjdsxc12@qq.com",
                "subject":"告知邮箱-温湿度异常",
                "body":email_body
            }if t_out_flag or h_out_flag else True,
        }
        return json(res)
    except Exception as e:
        return json({"error": str(e)}, status=500)
if __name__ == "__main__":
    app.run(host="0.0.0.0",port=9000)

# 作用是允许在字符串中直接嵌入变量或表达式，语法是用 {} 包裹变量名或计算式
def generate_email_body(sn,temperature,humidity,t_threshold,h_threshold):
    return (
        f"告警通知：\n\n"
        f"当前设备{sn}的温湿度数据超出正常范围。\n\n"
        f"设备温度：{temperature}°c\n"
        f"温度阈值:{t_threshold[0]}°c-{t_threshold[1]}°c\n\n"
        f"设备湿度：{humidity}%\n"
        f"湿度阈值:{h_threshold[0]}%-{h_threshold[1]}%\n\n"
        f"请尽快检查设备并采取相应措施。\n\n"
        f"时间:{datetime.datetime.now().isoformat()}"
    )

if __name__ == "__main__":
    app.run(host="0.0.0.0",port=9000)