系统中间件与云虚拟化实践

1. 基于阿里云云效Codeup的Git代码管理

1.1 实验目标与相关知识技能

• 了解Git常用指令及其基本原理，如：git push,git pull,git init,git add,git config,git branch,git commit等

• 了解一些常见的Linux系统命令，如：cd,cat,touch等

• 学会使用ssh方式访问仓库，减少不必要的交互以提高效率

1.2 实验步骤与对应成果展示

创建test.py文件，暂存至暂存区后观察文件状态，提交至本地仓库后观察文件状态，推送至远程仓库并观察文件状态

echo "import json" > test.py  #创建test.py
git add test.py  #暂存test.py
git status  #观察文件状态
git commit -m "feat(test.py):引入json"  #提交至本地仓库
git status  #观察文件状态
git push origin master  #推送至远程仓库
git status  #观察文件状态

至此，已完成暂存、提交、推送三个基本操作，本地代码已同步至云端Codeup

修改test.py文件后观察文件状态，暂存修改后的文件并观察文件状态，添加"import re"至test.py中，观察文件状态，提交至本地仓库后观察文件状态，推送至云端仓库后观察文件状态，查看历史提交记录

echo "import random" > test.py  #修正test.py
git status
git add test.py
git status
echo "import re" >> test.py
git status
git commit -a -m "feat(test.py)：引入random，re"
git status
git push
git status
git log  #git log用于查看历史提交记录

修改test.py，暂存文件test.py，取消暂存test.py，提交至本地仓库，撤销提交，再次提交至本地仓库，推送至云端仓库，本地回滚后再次强行推送至远程仓库，期间不断git status观察文件状态

echo "print('Hello world')" >> test.py
git status
git add test.py
git status
git restore --staged test.py  #将本地仓库HEAD指向的版本复制到暂存区
git status
git add test.py
git reset HEAD test.py
git commit -m "feat(test.py)：打印Hello World"
git status
git reset HEAD~  #HEAD~是HEAD的父节点，设置HEAD指向当前提交的上一次提交
git status
git commit -a -m "feat(test.py)：打印Hello World"
git push origin master
git reset HEAD~
git push -f origin master

取消对test.py的跟踪并恢复，与最近一次修改前的test.py进行比较，暂存test.py至暂存区，删除test.py，查看test.py的内容，从暂存区中恢复test.py，查看test.py内容，删除test.py，再强制删除已暂存的test.py，尝试恢复被删除的文件，查看test.py是否被恢复，从暂存区中恢复test.py，查看test.py的内容

Plain Text
git rm --cached test.py  #取消对test.py的跟踪
git status
git add test.py
git diff test.py
git status
git add test.py
git status
rm test.py  #删除test.py
cat test.py  #在终端中打印test.py的内容
git status
git restore test.py
cat test.py
git status
git rm test.py
git rm -f test.py
git status
git restore test.py
git restore --staged test.py
git status
git restore test.py
cat test.py

1.3 遇到的主要问题、解决思路和收获

该实验暂未遇到问题

2. 基于阿里云 ECS 与 ACR 的容器镜像管理

2.1 实验目标与相关知识技能

• 掌握ECS的基本操作，能较为熟练地使用Linux平台，如ECS远程连接，Xftp 7传输文件等

• 掌握Docker的常用命令，包括拉取镜像，启动容器，查看镜像，删除镜像，暂停容器等

• 学会撰写Dockerfile文件来构建镜像，并基于阿里云ACR完成镜像的pull和push

2.2 实验步骤与对应成果展示

使用docker pull拉取nginx镜像

1	docker pull nginx

使用docker images查看已拉取的镜像

1	docker images

使用docker run运行docker容器

1	docker run -d --name nginx -p 80:80 nginx

使用docker ps查看当前运行的容器信息

docker ps

此时可用本地浏览器访问该ECS服务器公网，即http://<ECS公网IP>以此验证容器是否正常运行

将Nginx服务的配置文件、日志文件及Web服务的根目录分别建立持久化映射，可以理解为让容器中的目录与宿主目录进行同步

mkdir -p /opt/docker/nginx/conf.d
mkdir -p /opt/docker/nginx/logs
mkdir -p /opt/docker/nginx/html
# -p --parents：如果路径中任意一级父目录不存在，则创建

docker cp nginx:/etc/nginx/nginx.conf /opt/docker/nginx/
docker cp nginx:/etc/nginx/conf.d /opt/docker/nginx/
docker cp nginx:/usr/share/nginx/html /opt/docker/nginx/

现在要想Nginx服务使用持久化存储的数据，需要先停止再删除当前运行的nginx容器

1
2
3

docker stop nginx
docker rm nginx
docker ps -a #观察是否删除成功

重新执行docker run，同时配置好相关参数

# -v, --volume list：挂在目录，<宿主目录>:<容器目录>
docker run -d --restart=always \
            --name nginx \
            -p 80:80 \
            -v /opt/docker/nginx/nginx.conf:/etc/nginx/nginx.conf \
            -v /opt/docker/nginx/html:/usr/share/nginx/html \
            -v /opt/docker/nginx/logs:/var/log/nginx \
            nginx

使用docker stop,docker start进行练习

1 2	docker stop nginx docker start nginx

使用curl命令获取网页内容

1	curl 127.0.0.1:80 #在ECS终端上输入，所以地址是127.0.0.1

使用docker exec命令进入容器。该命令定义为在容器内部运行一条指定命令，可以指定命令为Shell程序，如/bin/bash，配合参数-it，可以实现进入容器进行命令行交互式操作

docker exec -it nginx /bin/bash
# -i --interactive，保持交互模式
# -t --tty，分配一个伪终端（模拟终端）
# nginx，容器名称
# /bin/bash，待执行的命令，进入bash shell

进入容器后，在nginx服务的Web服务的根目录中创建test.html文件。由于容器本身不支持vim，因此用echo命令进行编写，最后用exit命令退出容器

cd /usr/share/nginx/html
echo "<h1>Hello,world</h1>" > test.html
exit
ls /opt/docker/nginx/html  #查看是否成功持久化储存到宿主系统
curl 127.0.0.1:80/test.html

除了curl命令，一样可以用浏览器访问http://<ECS公网地址>/test.html来验证

使用docker logs查看容器日志，可以通过此方法来判断容器是否正常工作

1	docker logs nginx

使用dcoker rmi删除不必要的容器镜像时，要先删除使用该镜像的容器，想删除容器又得先停止容器实例运行。故删除镜像的顺序如下：

docker stop nginx
docker rm nginx
docker rmi nginx
docker images  #判断镜像是否删除成功

构建Docker容器镜像可以通过编写并运行Dockerfile文件来实现，一般而言，Dockerfile的文件指令逻辑应按照以下模式建立：选择合适的基础镜像、安装基础工具与依赖、添加其他应用、清理缓存、声明镜像端口暴露情况、设置默认启动命令

# 使用Ubuntu作为基础镜像
FROM ubuntu:22.04
# 维护者信息
LABEL maintainer="<yud0u@qq.com>"

# 设置工作目录
WORKDIR /app

# 安装nginx及Python
RUN apt-get update && apt-get install -y \
    nginx \
    python3 \
    python3-distutils

# 清理apt软件包缓存
RUN rm -rf /var/lib/apt/lists/*

#声明暴露端口
EXPOSE 80

#设置启动命令
CMD["nginx", "-g", "daemon off;"]

使用docker build命令创建容器镜像（最好在Dockerfile同一目录下运行）

# -t, tag：镜像名称（REPOSITORY：TAG）
# npu：Nginx Python Ubuntu，自定义镜像名称
# .：PATH，执行命令的上下文路径，构造过程中可以引用该上下文中的任何文件
docker build -t npu .
docker images  #查看镜像是否创建成功

运行该镜像，检测Nginx服务；进入容器后检测Python解析能力

docker run -d --name npu -p 80:80 npu
curl 127.0.0.1
docker exec -it npu /bin/bash
(nginx)python3
exit()
(nginx)exit

将创建的npu镜像推送至ACR，在此之前先在ACR控制台创建镜像仓库npu

1
2
3

docker login --username=鱼豆YuDou crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com
docker tag [ImageId] crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/npu:[镜像版本号]
docker push crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/npu:[镜像版本号]

推送成功后可以在该镜像仓库中查看提交记录

从ACR中拉取容器镜像进行测试，需要先删除本地容器及其镜像

docker stop npu
docker rm npu
docker rmi npu
docker rmi crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/npu
docker images  #查看镜像是否删除成功

删除后再进行拉取操作，并进行测试

1
2
3

docker pull crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/npu:latest
docker run -d --name npu -p 80:80 crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/npu:latest
curl 127.0.0.1  #查看容器是否正常运行

2.4 遇到的主要问题、解决思路和收获

（✅）本实验遇到的最大问题为：docker拉取镜像超时。

为解决此问题，共尝试了三种方法：

（❌）在 /etc/docker/daemon.json 下配置docker镜像源，包括各种华为云等镜像源，配置方法参考以下链接 https://blog.csdn.net/weixin_50160384/article/details/139861337，但均以失败告终
（❌）在 /etc/docker/daemon.json 下配置阿里云镜像加速器，在ACR中获取加速器地址，并按如下方式配置，不添加其他镜像源，但效果并不稳定，有时可以拉取成功有时又不行

1
2
3

{
  "registry-mirrors":["https://***.mirror.aliyuncs.com"]
}

（✅）为ECS服务器配置代理，在网上阅读了许多文章后，我将具体方法总结在我的个人博客上： https://sweetyudou.github.io/2024/11/14/Docker/。简而言之，我使用clash成功为ECS服务器配置代理，并且成功做到了稳定拉取镜像，是一劳永逸的好方法

3. 基于阿里云ECS与ACR的Python微服务镜像构建、部署与接口访问

3.1 实验目标与相关知识技能

• 掌握Dockerfile构建容器镜像

• 理解HTTP协议基本概念，掌握HTTP调试工具的使用

• 理解网络服务API的概念，学会HTTP RESTful API的使用

• 理解并掌握Python Requests

3.2 实验步骤与对应成果展示

📂：exp_pyms_api_demo

在ACR中创建镜像仓库，以备后续的镜像上传

启动本实验提供的微服务范例，体验HTTP RESTful API接口，本服务范例在8000端口监听，启动命令为：

1
2

./exp_pyms_api_demo ./device.csv
# ./exp_pyms_api_demo <csv_file>,csv_file为存储设备信息的CSV文件路径。若该路径中存在CSV文件则读取信息，否则自动创建5条设备信息并生成CSV文件

此时访问http://<IP>:8000/docs将会看到以下页面：

此时当前目录创建device.csv文件：

1	ls -F #查看是否创建device.csv文件

Ctrl + C退出程序后，创建Dockerfile文件

# 使用Ubuntu 22.04 作为基础镜像
FROM ubuntu:22.04
LABEL maintainer="<yud0u@qq.com>"

# 指定工作目录为 /app
WORKDIR /app
#将主机文件系统中的 exp_pyms_api_demo 复制到镜像中
COPY exp_pyms_api_demo exp_pyms_api_demo

# 暴露服务端口 8000
EXPOSE 8000

# 指定容器的启动命令，devcie.csv 将存储在容器的/var/lib/exp_pyms_data/ 目录下，因此该目录应持久化
CMD ["/app/exp_pyms_api_demo", "/var/lib/exp_pyms_data/device.csv"]

使用docker build命令构建容器镜像，镜像名为exp_pyms_api_demo:1.0

1 2	docker build -t exp_pyms_api_demo:1.0 . docker images #查看镜像是否构建成功

将镜像上传至ACR镜像仓库

1
2
3

docker login --username=鱼豆YuDou crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com
docker tag [ImageId] crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/exp_pyms_api_demo:[镜像版本号]
docker push crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/exp_pyms_api_demo:[镜像版本号]

推送成功后，可在镜像仓库exp_pyms_api_demo的页面中查看刚刚推送的镜像

远程登录ECS并拉取镜像

1	docker pull crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/exp_pyms_api_demo:[镜像版本号]

在ECS工作目录创建持久化目录exp_pyms_data，用于存储容器内服务生成的业务数据文档，即设备信息数据，随后运行该镜像

mkdir exp_pyms_data
docker run -d --restart=always\
            --name exp_pyms_api_demo \
            -v ./exp_pyms_data/:/var/lib/exp_pyms_data/ \
            -p 8000:8000 \
            crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/exp_pyms_api_demo:1.0

使用docker logs查看服务是否成功启动

1 2	docker logs exp_pyms_api_demo ls exp_pyms_data/ #若有device.csv文件，则成功创建5条设备信息且服务成功运行

使用OpenAPI查看微服务接口，在浏览器中通过<ecs_ip>:8000/docs来访问Redocly文档地址

同理，本实验Swagger文档地址为<ecs_ip>:8000/docs/swagger

配置Python虚拟环境exp_venv

1
2
3

python3 -m venv exp_venv
source exp_venv/bin/activate
pip install requests

进入Python交互式界面进行初步验证

python3
import requests  #验证requests库是否可导入
requests.__version__  #查看requests版本
exit()

查看先前生成的device.csv文件内容，通过Requests库访问接口来获取设备信息

cat exp_pyms_data/device.csv
python3
import requests
resp = requests.get("http://<ecs_ip>:8000/v1/devices",json={"id":"6713124465"})  #通过id来检索设备并获取信息，该id为随机生成的例子
resp
resp.status_code  #查看状态码判断访问是否成功，访问成功的HTTP状态码为200
resp.text
resp.json()

同时还支持用“序号SN-类型”来检索设备

1
2
3

resp = requests.get("http://<ecs_ip>:8000/v1/devices", json={"sn": "XRmiI", "model": "Raspberry Pi 4"})  #该组合信息为随机生成的例子
resp.status_code
resp.json()

体验删除接口

1
2
3

resp = requests.delete("http://<ecs_ip>:8000/v1/devices", json={"ids":["3116641919", "94791787847", "1234567890"]})
resp.json()
cat exp_pyms_data/device.csv  #查看设备信息是否被删除

体验添加设备接口

resp = requests.post(
    "htpp://<ecs_ip>:8000/v1/devices",
    json={
        "name": "test-name",
        "type": "controller",
        "hardware":{
            "model": "test-model",
            "sn": "test-sn"
            },
            "software": {
                "version": "0.1",
                "last_update": "2023-08-06 20:00:00"
                },
                "nic": [
                    {
                        "type": "wifi",
                        "mac": "12:34:56:78:9a:bc",
                        "ipv4": "192.168.1.2"
                    }
                ],
                "status": "online"
            }
        )
resp.json()
cat exp_pyms_data/device.csv  #查看设备信息是否成功添加

体验更新设备信息接口

resp = requests.put(
    "http://127.0.0.1:8000/v1/devices",
    json={
        "id": "0922282528",
        "name": "new-name",
        "software": {
            "version": "0.5",
            "last_update": "2023-08-06 10:00:00"
        },
        "status": "offline"
    }
)
resp.json()
cat exp_pyms_data/device.csv  #查看是否成功更新设备信息

3.3 遇到的主要问题、解决思路和收获

该实验暂未遇到问题

4. 基于阿里云云效Flow的Python Web服务构建与部署

4.1 实验目标与相关知识技能

• 掌握Python虚拟环境的应用和操作

• 熟练运用阿里云云服务器ECS、阿里云云效代码管理Codeup与流水线Flow，以及阿里云容器镜像服务ACR

• 掌握Python项目部署的多种方式，初步体验流水线的应用模式

4.2 实验步骤与对应成果展示

📂：exp_pyms_demo

用Xftp 7把范例服务包上传至云端解压，并在ECS服务器上的虚拟环境部署范例服务

tar -xzvf exp_pyms_demo.tar.gz
cd exp_pyms_demo
python3 -m venv exp_venv
source exp_env/bin/activate
pip3 install -r requirements.txt

启动Python Web服务：

1	python3 server.py

使用本地浏览器访问<ECS_IP>:8000/，查看服务页面输出

确认无误后在终端输入Ctrl + C停止服务，退出虚拟环境venv并删除虚拟目录

1 2	deactivate rm -r exp_venv

在exp_pyms_demo目录下创建Dockerfile文件，将Python Web服务范例代码打包进容器镜像

# 基于Python 最新基础镜像
FROM python:3.10-slim-buster
LABEL maintainer="<yud0u@qq.com>"

# 工作目录
WORKDIR /app

# 安装Python库
COPY requirements.txt requirements.txt
RUN apt-get update && apt-get install -y \
    && pip3 install -r requirements.txt

# 清理 apt 软件包缓存
RUN rm -rf /var/lib/apt/lists/*

# 复制当前文件夹中全部文件到镜像中
COPY . .

# 暴露服务端口 8000
EXPOSE 8000

# 容器启动时执行命令
CMD [ "python3", "server.py"]

使用docker build命令构建镜像

1 2	docker build -t exp_pyms_demo . docker images #查询镜像是否构建成功

启动相关容器

1	docker run -d --restart=always --name exp_pyms_demo -p 8000:8000 exp_pyms_demo

同样在浏览器上访问http://<ECS_IP>:8000/，查看服务页面输出来验证。或者用curl命令来验证

1	curl http://127.0.0.l:8000

验证成功后停止、删除容器后，删除镜像

1
2
3

docker stop exp_pyms_demo
docker rm exp_pyms_demo
docker rmi exp_pyms_demo

本地下载范例服务包后，进行本地仓库初始化并关联远程仓库,创建venv分支为后续实验做准备

git init
git remote add origin git@codeup.aliyun.com:66e7c39e4244e4202214531d/exp_pyms_demo.git
git pull origin master
git add .
git commit -m "init"
git push -u origin master
git chechout -b venv  #创建venv分支并切换
git push --set-upstream origin venv  #创建远程仓库，与本地仓库关联

配置云效Flow以支持ECS虚拟环境部署，选择代码源为Codeup的代码仓库exp_pyms_demo，默认分支为venv，将ECS服务器配置到主机部署中，运行流水线

此时可以通过curl命令测试服务接口是否可访问，或通过浏览器访问<ECS_IP>:8000/

1	curl http://127.0.0.1:8000

修改devices.csv，添加一条设备信息，并推送至远程仓库触发流水线执行，用浏览器重新访问

可以看到已显示添加的设备信息，关闭范例服务

将之前准备好的Dockerfile文件用于在流水线中制作承载范例服务的Docker镜像

1
2
3

git add Dockerfile
git commit -m "docs: Add Dockerfile"
git push --set-upsstream origin docker

重新配置好流水线源后运行流水线，改用Docker部署

根据「Python代码扫码阶段」所提示的信息，修改server.py存在的格式问题后提交、推送

4.3 遇到的主要问题、解决思路和收获

(✅）本实验遇到的最大问题为：Dockerfile中拉取python:3.10-slim-buster超时

（❌）根据该页面的智能排查，我尝试从阿里云ACR制品中心中拉取alinux3/python来替代，后续出现一系列如：apt-get命令不存在，需替换成apk等问题，最后发现是操作系统之间不兼容，下载的python是基于alinux3的

（✅）咨询了阿里云在线客服后，把「Python镜像构建」配置中的构建集群改为「云效中国香港构建集群」，即可顺利拉取

5. 基于阿里云函数计算的简单邮件发送服务设计与体验

5.1 实验目标与相关知识技能

• 理解并掌握函数计算FC的设计部署过程

• 理解并体验函数计算FC的弹性伸缩能力

5.2 实验步骤与对应成果展示

创建自定义公共层，提供Python Sanic依赖，以供后续实验进行。兼容运行时选择Debian 10，构建环境选择Python 3.10，requirements.txt文件中输入sanic

创建Web函数fun-alarm-email-send，，基于Sanic框架编写接口代码，构建并部署告警邮件发送接口

# -*- coding: utf-8 -*-
from sanic import Sanic
from sanic.response import json
from smtplib import SMTP
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
app = Sanic("EmailSender")

# 邮件配置 需要自行修改。邮箱授权码(password)的获取，自行前往对应
EMAIL_CONFIG = {
    "host": "smtp.example.com",
    "port": 587,
    "username": "your-email@example.com",
    "password": "your-password",
    "sender": "your-email@example.com"
}

def send_email(recipient, subject, body):
    # 创建邮件对象
    msg = MIMEMultipart()
    msg['From'] = EMAIL_CONFIG["sender"]
    msg['To'] = recipient
    msg['Subject'] = subject
    # 添加邮件正文
    msg.attach(MIMEText(body, 'plain'))
    # 连接SMTP服务器
    server = SMTP(EMAIL_CONFIG["host"], EMAIL_CONFIG["port"])
    server.starttls()  # 启动TLS加密
    server.login(EMAIL_CONFIG["username"], EMAIL_CONFIG["password"])
    # 发送邮件
    server.send_message(msg)
    # 关闭连接
    server.quit()

@app.route("/send", methods=["POST"])
def send_email_route(request):
    data = request.json
    recipient = data.get("recipient")
    subject = data.get("subject")
    body = data.get("body")
    if not all([recipient, subject, body]):
        return json({"error": "Missing required fields"})
    try:
        send_email(recipient, subject, body)
        return json({"message": "Email sent successfully"})
    except Exception as e:
        return json({"error": str(e)}, status=500)

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

代码修改后，删除现有的官方公共层Flask，并添加自定义创建的sanic-custom-layer层后部署，最后部署代码。利用Apifox工具编写对邮件发送接口发起HTTP接口

点进函数详情页面的配置-运行时按钮，修改单实例并发度为2，最后部署。随后在Apifox上点击自动化测试按钮新增测试场景，填写请求信息后，将线程数设置为10，即同时并发执行的线程数一共10个。监控函数详情页面的实例列表

5.3 遇到的主要问题、解决思路和收获

该实验暂未遇到问题

6. 基于阿里云函数计算的云工作流CloudFlow设计与体验

6.1 实验目标与相关知识技能

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

6.2 实验步骤与对应成果展示

在https://ram.console.aliyun.com/roles中创建角色，分配的权限包括：AliyunFCFullAccess、AliyunFnFFullAccess、AliyunEventBridgeFullAccess

在https://eventbridge.console.aliyun.com/overview中一键授权，对EventBridge进行授权

构建名为fun-temperature-and-humidity-data-upload的FC函数

# -*- coding: utf-8 -*-
import datetime
from sanic import Sanic
from sanic.response import json

app = Sanic("MyApp")

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

@app.route("/upload", methods=["POST"])
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 required fields"}, status=400)

        # 判断温湿度是否超出阈值
        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": "synx@example.com",
                "subject": "告警邮件 - 温湿度异常",
                "body": email_body
            } if t_out_flag or h_out_flag else None
        }
        return json(res)
    except Exception as e:
        return json({"error": str(e)}, status=500)

def generate_email_body(sn, temperature, humidity, t_threshold, h_threshold):
    return (
        "告警通知：\n\n"
        "当前设备({sn})的温湿度数据超出正常范围。\n\n"
        "设备温度：{temperature}°C\n"
        "温度阈值：{t_threshold[0]}°C - {t_threshold[1]}°C\n"
        "设备湿度：{humidity}%\n"
        "湿度阈值：{h_threshold[0]}% - {h_threshold[1]}%\n"
        "请尽快检查设备并采取相应措施。\n\n"
        "时间：{datetime.datetime.now().isoformat()}"
    )

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

此时，FC函数列表中应该有两个函数

创建云工作流，地域需与FC所在地域保持一致。采用快速模式来设置工作流，点击CloudFlow Studio按钮回到画布编辑页面，删除工作流默认创建的Hello World，并从左侧操作拖拽InvokeFunction任务状态到画布上，命名为fun-temperature-and-humidity-data-upload，右侧基本配置中切换YAML编辑，在YAML框内添加body.$: $Input.body，点击保存。在输出配置中，勾选使用JsonPath选择部分参数，填入$Output.Body，点击保存。

再拖拽一个InvokeFunction任务，命名为fun-alarm-email-end，配置YAML数据负载时，填写body.$: $Input.data

从左侧拖拽Choice放置在两个InvokeFunction之间，再拖拽Succeed补全步骤，Choice默认规则的下个状态设置为Succeed，自定义 #1卡片添加条件$Input.status == 1，并设置下个状态为fun-alarm-email-send，点击保存

修改fun-alarm-email-sendFC代码，以适配云工作流调用

Python
# 源代码：
@app.route("/send", methods=["POST"])
# 修改为：
@app.route("/send", methods=["POST"], name='send')
@app.route("/invoke", methods=["POST"], name='invoke')

修改fun-temperature-and-humidity-data-uploadFC代码

# 源代码：
@app.route("/upload", methods=["POST"])
# 修改为：
@app.route("/upload", methods=["POST"], name='upload')
@app.route("/invoke", methods=["POST"], name='invoke')

返回工作流详情界面，点击工作流调度标签页，创建工作流调度并选择HTTP/HTTPS触发，请求类型选择HTTPS，请求方法为POST，返回工作流详情页面后点击详情按钮，复制公网访问地址

使用Apifox对云工作流公网发起POST请求

返回工作流详情页面，点击执行记录标签，将出现一条执行记录

点击详情可查看详细的执行流程

修改请求的数据，将temperature改为40，模拟设备出现异常情况。发起请求后查看详细执行流程

6.3 遇到的主要问题、解决思路和收获

本实验仅遇到一个乌龙，花了一个多小时才debug出来：即在flow-temperature-and-humidity-data-upload的YAML编辑中，错把$Input.body填成$Input.Body，填写了大写的B。导致后续用Apifox发起请求时返回结果与范例不同。调用云工作流内部的测试功能，观察每个InvokeFunction的输入与输出后，意识到请求数据从未进入过流程当中，于是问题锁定在第一个InvokeFunction中，在研究数据如何导入的过程中，发现数据是根据$Input.body导入的，最终发现并解决了这个乌龙

7. 基于阿里云函数计算的简单邮件发送服务之数据库访问中间件

7.1 实验目标与相关知识技能

• 掌握基于PyMySQL使用原生SQL开展数据库表访问的基本操作

• 掌握基于ORM框架的SQLAlchemy开展数据库表访问

7.2 实验步骤与对应成果展示

进入PolarDB的控制台，创建集群后，为集群配置白名单，允许所有IP访问(0.0.0.0/0)

在基本信息中找到数据库连接，复制私网地址以供后续实验使用

查看当前专有网络VPC和交换机vSwitch信息，保证后续实验产品共用同一套VPC和vSwitch

构建FC自定义公共层，提供Python sqlalchemy依赖，并为fun-alarm-email-send函数配置相同的VPC和vSwitch

使用阿里云提供的云数据库统一控制台(https://dmslab.aliyun.com/)对PolarDB进行操作，登录实例后进入管理页面，点击创建库按钮新建db_message数据库，字符集选择utf8mb4%E5%AF%B9PolarDB%E8%BF%9B%E8%A1%8C%E6%93%8D%E4%BD%9C%EF%BC%8C%E7%99%BB%E5%BD%95%E5%AE%9E%E4%BE%8B%E5%90%8E%E8%BF%9B%E5%85%A5%E7%AE%A1%E7%90%86%E9%A1%B5%E9%9D%A2%EF%BC%8C%E7%82%B9%E5%87%BB%60%E5%88%9B%E5%BB%BA%E5%BA%93%60%E6%8C%89%E9%92%AE%E6%96%B0%E5%BB%BA%60db_message%60%E6%95%B0%E6%8D%AE%E5%BA%93%EF%BC%8C%E5%AD%97%E7%AC%A6%E9%9B%86%E9%80%89%E6%8B%A9utf8mb4)

打开db_message的SQL控制台，使用以下SQL语句创建tbl_config表结构

CREATE TABLE `tbl_config` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '配置记录的唯一标识符',
  `host` varchar(255) NOT NULL COMMENT '邮件服务器的主机地址',
  `post` int(5) NOT NULL COMMENT '邮件服务器的端口号',
  `username` varchar(100) NOT NULL COMMENT '登录邮件服务器的用户名，示例：1464935327@qq.com',
  `password` varchar(100) NOT NULL COMMENT '登录邮件服务器的授权码',
  `sender` varchar(255) NOT NULL COMMENT '邮件发送人的地址，示例：1464935327@qq.com',
  `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '记录创建时间',
  `update_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '记录最后更新时间',
  PRIMARY KEY(`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='邮件配置表';

SELECT * FROM `tbl_config`
 LIMIT 20;

重写fun-alarm-email-send函数，新建email_config_service.py，代码如下：

import json
from typing import Any, Dict
import pymysql
from custom_json_encoder import DateTimeEncoder

def get_db_connection():
    return pymysql.connect(
        host='pc-bp1f07b8tpl06q3qu.mysql.polardb.rds.aliyuncs.com',
        port=3306,
        user='yudou',
        password='ZYDzyd917917',
        db='db_message',
        charset='utf8mb4',
    )

async def create_config(data) -> Dict[str, Any]:
    try:
        with get_db_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute("""
                    INSERT INTO tbl_config (
                        host,
                        port,
                        username,
                        password,
                        sender)
                        VALUES (%s, %s, %s, %s, %s)
                """, (data['host'], data['port'], data['username'], data['password'], data['sender']))
                connection.commit()
        return {"message": "Config created successfully"}
    except Exception as e:
        connection.rollback()
        return {"error": str(e)}

async def read_config(data) -> Dict[str, Any]:
    try:
        with get_db_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute("SELECT * FROM tbl_config LIMIT 1")
                result = cursor.fetchone()
                if result:
                    config = {
                        "id": result[0],
                        "host": result[1],
                        "port": result[2],
                        "username": result[3],
                        "password": result[4],
                        "sender": result[5],
                        "create_time": result[6],
                        "update_time": result[7],
                    }
                    return json.loads(json.dumps(config, cls=DateTimeEncoder))
                else:
                    return {"error": "No configuration found"}
    except Exception as e:
        return {"error": str(e)}

async def update_config(data) -> Dict[str, Any]:
    try:
        with get_db_connection as connection:
            with connection.cursor() as cursor:
                if not data.get("id"):
                    raise ValueError("id is required")

                sql = "UPDATE tbl_config SET"
                params = []

                for key in ['host', 'port', 'username', 'password', 'sender']:
                    if data.get("key"):
                        sql += f" {key} = %s,"
                        params.append(data[key])

                sql = sql.rstrip(",") + " WHERE id = %s"
                params.append(data["id"])

                if cursor.execute(sql, tuple(params)) == 0:
                    raise ValueError("Config not found")
                connection.commit()
            return {"message": "Config updated successfully"}
    except Exception as e:
        connection.rollback()
        return {"error": str(e)}

async def delete_config(data) -> Dict[str, Any]:
    try:
        with get_db_connection() as connection:
            with connection.cursor() as cursor:
                if not data.get("id"):
                    raise ValueError("id is required")

                if cursor.execute("DELETE FROM tbl_config WHERE id = %s", (data["id"],)) == 0:
                    raise ValueError("Config not found")
                connection.commit()
        return {"message": "Config deleted successfully"}
    except Exception as e:
        connection.rollback()
        return {"error": str(e)}

新建custom_json_encoder.py，代码如下：

import json
from datetime import datetime

class DateTimeEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetme):
            return obj.isoformat()
        return super().default(obj)

修改app.py代码

import json as std_json
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from smtplib import SMTP

from sanic import Sanic,response
from sanic.response import json

from email_config_service import (create_config,
                                  delete_config,
                                  read_config,
                                  update_config)

app = Sanic("EmailSender")

async def send_email(data):
        email_config = await read_config(None)

        msg = MIMEMultipart()
        msg['From'] = email_config["sender"]
        msg['To'] = data.get("recipient")
        msg['Subject'] = data.get("subject")

        msg.attach(MIMEText(data.get("body"), 'plain'))

        server = SMTP(email_config["host"], email_config["port"])
        server.starttls()
        server.login(email_config["username"],email_config["password"])

        server.send_message(msg)

        server.quit()
        return {"message": "Email sent successfully"}

@app.route("/invoke", methods=["POST"])

async def send_email_route(request):
        action = request.json.get("action")
        data = request.json.get("data",{})

        actions = {
                "create_config": create_config,
                "read_config": read_config,
                "update_config": update_config,
                "delete_config": delete_config,
                "send_email": send_email
        }

        func = actions.get(action)
        if func:
                return json(await func(data))
        else:
                return response.json({"error": "Invalid action"},status=400)

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

使用Apifox进行测试新增数据库数据

打开db_message库中的tbl_config表，可发现数据已存储

测试告警邮件

基于sqlalchemy重写email_config_service.py

import json
import urllib.parse
from typing import Any, Dict

from sqlalchemy import Column, DateTime, Integer, String, create_engine, func
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from custom_json_encoder import DateTimeEncoder

Base = declarative_base()

class Config(Base):
    __tablename__ = 'tbl_config'
    id = Column(Integer, primary_key=True, autoincrement=True)
    host = Column(String(255))
    port = Column(Integer)
    username = Column(String(255))
    password = Column(String(255))
    sender = Column(String(255))
    create_time = Column(DateTime, server_default=func.now())
    update_time = Column(DateTime, server_default=func.now(), onupdate=func.now())

def get_db_connection():
    DB_HOST = 'pc-bp1f07b8tpl06q3qu.mysql.polardb.rds.aliyuncs.com'
    DB_PORT = 3306
    DB_USER = 'yudou'
    DB_PASSWORD = 'ZYDzyd917917'
    DB_NAME = 'db_message'
    encoded_password = urllib.parse.quote_plus(DB_PASSWORD)
    DATABASE_URL = f'mysql+pymysql://{DB_USER}:{encoded_password}@{DB_HOST}:{DB_PORT}/{DB_NAME}'
    engine = create_engine(DATABASE_URL, echo=True)
    Session = sessionmaker(bind=engine)
    return Session()

async def create_config(data) -> Dict[str, Any]:
    """创建配置"""
    with get_db_connection() as session:
        try:
            config = Config(
                host=data['host'],
                port=data['port'],
                username=data['username'],
                password=data['password'],
                sender=data['sender']
            )
            session.add(config)
            session.commit()
            return {"message": "Config created successfully"}
        except SQLAlchemyError as e:
            session.rollback()
            return {"error": str(e)}

async def read_config(data) -> Dict[str, Any]:
    """读取配置，此处仅做最简单的查询，实际应用中需要根据业务需求进行查询"""
    with get_db_connection() as session:
        try:
            config = session.query(Config).first()
            if config:
                config_dict = {
                    "id": config.id,
                    "host": config.host,
                    "port": config.port,
                    "username": config.username,
                    "password": config.password,
                    "sender": config.sender,
                    "create_time": config.create_time,
                    "update_time": config.update_time
                }
                encoded_result = json.dumps(config_dict, cls=DateTimeEncoder)
                return json.loads(encoded_result)
            else:
                return {"error": "No configuration found"}
        except SQLAlchemyError as e:
            session.rollback()
            return {"error": str(e)}

async def update_config(data) -> Dict[str, Any]:
    """更新配置"""
    with get_db_connection() as session:
        try:
            config = Config(
                host=data['host'],
                port=data['port'],
                username=data['username'],
                password=data['password'],
                sender=data['sender']
            )
            session.add(config)
            session.commit()
            return {"message": "Config updated successfully"}
        except SQLAlchemyError as e:
            session.rollback()
            return {"error": str(e)}

async def delete_config(data) -> Dict[str, Any]:
    """删除配置"""
    with get_db_connection() as session:
        try:
            config_id = data.get('id', None)
            if not config_id:
                raise ValueError("id is required")
            config = session.query(Config).filter(Config.id == config_id).first()
            if not config:
                raise ValueError("Config not found")
            session.delete(config)
            session.commit()
            return {"message": "Config deleted successfully"}
        except SQLAlchemyError as e:
            session.rollback()
            return {"error": str(e)}

同上进行测试，结果应该不变

7.3 遇到的主要问题、解决思路和收获

该实验暂未遇到问题

8. 基于阿里云Serverless应用引擎的服务部署迁移

8.1 实验目标与相关知识技能

• 熟练运用云效Codeup代码仓库的使用

• 理解并掌握阿里云Serverless应用引擎部署应用的方法

8.2 实验步骤与对应成果展示

📂：user_admin

📂：device_control

将以上用户管理服务代码与设备管理服务代码上传到云效Codeup

修订user-admin中application.yml的代码。(12~~14行)中MySQL数据库配置更改为PolarDB MySQL的内网访问地址及对应的用户名密码。(42~~44行)中Redis数据库配置为Redis数据库配置信息

打开device的代码，修改server-config.ini文件中MySQL数据库配置为PolarDB MySQL的内网访问地址及其对应的用户名密码。callback_url替换为CloudFlow的内网触发地址

进入阿里云DMS控制台，在PolarDB MySQL数据库中创建db_user_admin数据库，点击左侧常用功能-数据导入，选择批量数据导入，数据库选择db_user_admin，文件类型为SQL脚本，最后上传deviced_control的init.sql文件

修改fun-temperature-and-humidity-data-uploadFC函数

# 原来温湿度数据上报FC的res数据结构如下
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": "1464935327@qq.com",
        "subject": "告警邮件 - 温湿度异常",
        "body": "email_body"
    }
}
if t_out_flag or h_out_flag else None

# 应该修订为以下格式
res = {
    "status": 1 if t_out_flag or h_out_flag else 0,
    "message": "异常" if t_out_flag or h_out_flag else "正常",
    "data": {
        "action": "send_email",
        "data": {
            # 需要自定转换成收件人邮箱地址
            "recipient": "1464935327@qq.com",
            "subject": "告警邮件 - 温湿度异常",
            "body": "email_body"
        }
    }
}
if t_out_flag or h_out_flag else None

进入SAE控制台创建Web应用，VPC与vSwitch不变，代码仓库类型选择Codeup，仓库选择serverless/user-admin

启动命令如下设置：

1	java -jar target/user_admin-2024.jar

使用Apifox对User-Admin登录接口(POST /login/account)发起请求

请求实例：

POST https://cn-hanger-admin-hvryfblqme.cn-hangzhou.sae.run/login/account
{
    "account": "yudou",
    "password": "123456"
}

响应示例：

{
    "status": 1,
    "message": "操作执行成功",
    "data": "65086510285720102"
}

同理，基于SAE部署Device Control服务。设置启动命令为如下命令：

1	python server.py

使用Apifox对Device-Control的设备添加接口(POST /devices)发起请求

请求示例：

POST https://cn-hang-control-rcuipmgcxu.cn-hangzhou.sae.run
[
    {
        "name": "温湿度设备",
        "type": "null",
        "sn": "SNTAH",
        "passwd": "123456"
    }
]

响应示例：

{
    "status": 1,
    "message": "设备信息添加成功！"
}

📂：device_client

以上为模拟设备客户端代码示例，设备客户端用于模拟真实设备，构造温湿度数据并定期通过WebSocket通道上传数据至Device-Control

# 第一个参数为Device - Control应用的地址(注意：不需要https://，只需要域名)
# 第二个参数是前面步骤创建的设备sn
# 第三个参数是该设备的密码
(venv)$ python client.py ws://<your_sae_domain>/devices/auth/ws SNT AH 123456

INFO:root:成功连接到 ws://<your_sae_domain>/devices/auth/ws? sn=SNT AH&passwd=123456

REC:身份验证成功
REC:{"sn":"SNT AH","temperature":26.15,"humidity":31.97}
REC:{"sn":"SNT AH","temperature":25.25,"humidity":31.43}
REC:{"sn":"SNT AH","temperature":26.85,"humidity":30.08}
REC:{"sn":"SNT AH","temperature":35.8,"humidity":27.67}
REC:{"sn":"SNT AH","temperature":25.46,"humidity":30.67}

向Device-Control发送温湿度数据后，可以自行查看CloudFlow的云效情况，若CloudFlow正常工作，在温湿度超出阈值时将向指定的管理员邮箱发生告警邮件

8.3 遇到的主要问题、解决思路和收获

（❌）该实验并未完成，在测试两个SAE服务时遇到无法解决的bug

上图为user-admin服务的日志，显示无法连接到PolarDB MySQL，但我反复确认各个阿里云产品的VPC和vSwitch都是一样的

以上为Apifox测试时的报错，第一反应是Redis和PolarDB MySQL地址填写错误，检查Codeup内代码发现并未出错。再依次检查各个云产品的VPC，vSwitch，以及白名单，也未修复该bug。最后尝试删除user-adminWeb服务重做，还是出现相同bug

9. 基于Nacos的服务注册与配置部署

9.1 实验目标与相关知识技能

• 理解并掌握微服务系统中，服务注册与发现机制的基本原理，相关中间件的系统定位和功能

• 理解并掌握服务注册与发现中间件Nacos的具体功能与应用模式

9.2 实验步骤与对应成果展示

📂：user_admin

📂：device_control

创建ACR镜像仓库，使用Docker拉取Nacos镜像，并上传至阿里云容器镜像仓库ACR

docker pull nacos

docker login --username=鱼豆YuDou crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com
docker tag [ImageId] crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/nacos:[镜像版本号]
docker push crpi-n4sctgr4gzd05xye.cn-hangzhou.personal.cr.aliyuncs.com/exp_yd/nacos:[镜像版本号]

在SAE控制台创建微服务，应用名称为Nacos，保证VPC，vSwitch，安全组与先前实验保持一致，添加环境变量MODE=standalone，部署Nacos

开通CLB负载均衡后，用浏览器访问http://<CLB公网IP>:8848/nacos，即可观察Nacos UI界面

在user-admin包中找到pom.xml，添加Nacos的依赖坐标

<dependencies>
    ...... (省略之前的依赖)
    <dependency>
        <groupId>com.alibaba.boot</groupId>
        <artifactId>nacos-discovery-spring-boot-starter</artifactId>
        <version>0.3.0-RC</version>
    </dependency>
</dependencies>

修改src/main/resources/application.yml文件，添加Nacos配置，实现自动注册

nacos:
  discovery:
    server-addr: 172.28.97.37:8848
    auto-register: true
    register:
      service-name: user-admin
      ip: https://cn-hanger-admin-hvryfblqme.cn-hangzhou-vpc.sae.run
      ephemeral: false
      healthy: true

修改后推送代码至Codeup触发自动部署，进入Nacos UI界面可以看到user-admin服务已注册

修改Device-Control中的server.py代码，实现应用启动时自动向Nacos注册

@app.listener('before_server_start')
async def register_nacos(app, loop):
    try:
        import nacos

        # 自行替换为 Nacos 内网ip地址
        SERVER_ADDRESSES = "http://172.28.97.37:8848"
        # Nacos注册的信息
        SERVICE_NAME = "device-control"
        # 自行替换为SAE中获取的内网访问地址，只需要域名，无需携带https://
        IP = "https://cn-hang-control-rcuipmgcxu.cn-hangzhou.sae.run"
        PORT = 9000

        client = nacos.NacosClient(SERVER_ADDRESSES)
        # 向 Nacos 注册实例
        client.add_naming_instance(SERVICE_NAME, IP, PORT, ephemeral=False, healthy=True)
        print(f"向 Nacos 注册成功，注册信息为: {SERVICE_NAME}, {IP}, {PORT}")
    except Exception as e:
        raise RuntimeError(f"向 Nacos 注册失败，错误信息: {e}")

同时修改requirements.txt文件，添加一行nacos-python-sdk依赖

sanic~=24.6.0
pymysql~=1.1.1
websockets~=13.0.1
requests~=2.32.3
nacos-python-sdk==0.2

修改代码后推送至Codeup，触发自动部署。打开Nacos UI界面即可验证是否注册

在部署Nacos SAE应用实例的终端环境下执行以下命令，实现邮件发送FC函数计算的静态注册

curl --location --request POST 'http://127.0.0.1:8848/nacos/v2/cs/config?group=DEFAULT_GROUP&dataId=serverless.fc.address.email_send&content=https://fun-alaail-send-ylnkfxeuva.cn-hangzhou-vpc.fcapp.run'

配置完成后观察Nacos UI界面，可发现send_email配置已成功向Nacos发布

同理，将CloudFlow工作流的触发接口以静态注册的方式进行

curl --location --request POST
'http://127.0.0.1:8848/nacos/v2/cs/config?
group=DEFAULT_GROUP&
dataId=serverless.fc.address.cloudflare&
content=https://1237899087648526.eventbridge.cn-hangzhou-vpc.aliyuncs.com/webhook/putEvents?token=1ced1dfd257949609f8cd8861b0a914bba5389a992f04ac8a17616ba7e360318f463ec2f6fa14a79b4406e8ed277e4eea86fe72c7e384b82a8fa3825705a6a6d'

构建用户服务FC，同样配置相同的VPC、vSwitch，代码如下：

from sanic import Sanic
from sanic.response import json, text
import requests

app = Sanic("UserAdminFC")

# Nacos配置
NACOS_URL = "http://172.28.97.37:8848"  # 自行替换为你的nacos内网地址
SERVICE_NAME = "user-admin"

# 在全局变量中存储user - admin服务的实例URL信息
service_instance_url = None

def get_service_instance_url():
        global service_instance_url
        if service_instance_url:
                return service_instance_url

        params = {
                "serviceName": SERVICE_NAME,
        }
        response = requests.get(
                f"{NACOS_URL}/nacos/v2/ns/instance/list",
        params=params)
        if response.status_code == 200:
                data = data = response.json().get("data",
                {'hosts': []}).get('hosts')
                if data and data[0]:
                        service_instance_url = data[0].get('ip')
                        return service_instance_url
        return None

@app.route("/register/account",methods=["POST"])
async def register_account(request):
        """用户注册"""
        instance_url = get_service_instance_url()
        if instance_url:
                url = f"http://{instance_url}/register/account"
                response = requests.post(url, json=request.json)
                return json(response.json())
        else:
                return text("Service instance not found", status=404)

@app.listener('before_server_start')
async def setup(app, loop):
        """在启动时获取服务实例信息"""
        instance_url = get_service_instance_url()
        if instance_url is None:
                raise RuntimeError("Service instance not found")
        print("成功获取到服务实例地址：" + instance_url)

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

使用Apifox对User-Admin FC的用户注册接口(POST /register/account)发起请求，接口请求示例如下：

POST https://useradminfc-zovpsrnfjk.cn-hangzhou.fcapp.run/register/account
{
    "account": "yudou",
    "password": "123456"
}

响应示例：

{
    "status": 1,
    "message": "操作执行成功",
    "data": "65086872090576958"
}

构建设备管理FC代码：

import requests
from sanic import Sanic
from sanic.response import json, text

app = Sanic("DeviceControlFC")

# Nacos配置
NACOS_URL = "http://172.28.97.37:8848"  # 自行替换为你的nacos内网地址
SERVICE_NAME = "device-control"  # 服务名称

# 在全局变量中存储device - control服务的实例URL信息
service_instance_url = None

def get_service_instance_url():
    global service_instance_url
    if service_instance_url:
        return service_instance_url

    params = {
        "serviceName": SERVICE_NAME,
    }
    response = requests.get(
        f"{NACOS_URL}/nacos/v2/ns/instance/list",
        params=params)
    if response.status_code == 200:
        data = response.json().get("data", {'hosts': []}).get('hosts')
        if data and data[0]:
            service_instance_url = data[0].get('ip')
            return service_instance_url
    return None

@app.post("/devices")
async def create_device(request):
    """设备添加"""
    instance_url = get_service_instance_url()
    if instance_url:
        url = f"http://{instance_url}/devices"
        response = requests.post(url, json=request.json)
        return json(response.json())
    else:
        return text("Service instance not found", status=404)

@app.get("/devices", ignore_body=False)
async def query_device(request):
    """设备查询"""
    instance_url = get_service_instance_url()
    if instance_url:
        url = f"http://{instance_url}/devices"
        headers = {
            "Authorization": request.headers.get("Authorization")
        }
        response = requests.get(url, json=request.json, headers=headers)
        return json(response.json())
    else:
        return text("Service instance not found", status=404)

@app.listener('before_server_start')
async def setup(app, loop):
    """在启动时获取服务实例信息"""
    instance_url = get_service_instance_url()
    if instance_url is None:
        raise RuntimeError("Service instance not found")
    print("成功获取到服务实例地址：" + instance_url)

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

同理，用Apifox对Device-Control FC的设备添加接口(POST /devices)发起请求，接口请求示例如下：

POST https://device-ontrolfc-rrubtancws.cn-hangzhou.fcapp.run/devices
[
    {
        "name": "温湿度设备",
        "type": "null",
        "sn": "SNTAH",
        "passwd": "123456"
    }
]

接口响应示例：

{
    "status": 1,
    "message": "获取设备添加成功！"
}

对设备查询接口(GET /devices)发起请求，接口请求示例如下：

1	GET https://device-ontrolfc-rrubtancws.cn-hangzhou.fcapp.run/devices

接口响应示例：

{
    "status": 1,
    "message": "获取设备信息成功！",
    "data": [
        {
            "id": "1",
            "name": "温湿度设备",
            "type": "null",
            "sn": "SNTAH",
            "passwd": "123456"
        },
        ...
    ]
}

构建温湿度业务FC，示例代码如下：

import requests
from sanic import Sanic, response
from sanic.request import Request

app = Sanic("BizFC")

# Nacos配置
NACOS_URL = "http://172.28.97.37:8848"

def get_services_from_nacos():
    """从Nacos获取服务列表"""
    response = requests.get(f"{NACOS_URL}/nacos/v2/ns/service/list")
    data = response.json()
    return data['data']['services']

def get_instance_from_nacos(service_name):
    """从Nacos获取指定服务的实例IP"""
    response = requests.get(f"{NACOS_URL}/nacos/v2/ns/instance/list?serviceName={service_name}")
    data = response.json()
    hosts = data['data'].get('hosts', [])
    if len(hosts) > 0:
        return data['data']['hosts'][0]['ip']
    return ""

def get_config_from_nacos(data_id, group):
    """从Nacos获取指定配置"""
    response = requests.get(f"{NACOS_URL}/nacos/v2/cs/config?dataId={data_id}&group={group}")
    data = response.json()
    if data['data']:
        return data['data']
    return ""

@app.get("/getServices")
async def get_services(request: Request):
    """获取服务及其实例IP，并返回JSON格式的服务映射"""
    services = get_services_from_nacos()
    service_map = {}
    for service in services:
        instance_ip = get_instance_from_nacos(service)
        service_map[service.replace('-', '_')] = instance_ip
    # 从静态配置中读取
    static_services = {
        "message_url": get_config_from_nacos("serverless.fc.address.email_send", "DEFAULT_GROUP"),
        "cloudflow_url": get_config_from_nacos("serverless.fc.address.cloudflow", "DEFAULT_GROUP")
    }
    service_map.update(static_services)
    return response.json(service_map)

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

使用Apifox对Biz FC的获取的所有服务地址接口(GET /getServices)发起请求，接口请求示例如下：

1	GET https://bizfc-cpklnjfjoq.cn-hangzhou.fcapp.run/getServices

接口响应示例：

{
    "device_control": "https://cn-hang-control-rcuipmgcxu.cn-hangzhou.sae.run",
    "user_admin": "https://cn-hanger-admin-hvryfblqme.cn-hangzhou.sae.run",
    "message_url": "config data not exist",
    "cloudflow_url": "https://1237899087648526.eventbridge.cn-hangzhou.aliyuncs.com/webhook/putEvents?token=1ced1dfd257949609f8cd8861b0a914bba5389a992f04ac8a17616ba7e360318f463ec2f6fa14a79b4406e8ed277e4eea86fe72c7e384b82a8fa3825705a6a6d"
}