使用云函数进行分布式并行计算

Michael阿明

发布于 2023-12-08 15:01:38

1770

发布于 2023-12-08 15:01:38

文章被收录于专栏：Michael阿明学习之路Michael阿明学习之路

文章目录

1. 云函数概念
2. web云函数
3. 部署+发布新版本
4. 测试
5. 事件函数
6. 使用场景

1. 云函数概念

云函数是一段运行在云端的代码，无需管理服务器，在开发工具内编写、一键上传部署即可运行后端代码。

云函数的原理是基于事件驱动，当指定事件发生时，自动触发云函数执行。

云函数平台根据请求量自动弹性伸缩，事后仅按量计费，无需担心性能瓶颈和闲置成本。

云函数提供API网关触发、定时触发、消息队列触发等多种触发方式，方便用户灵活调用。

云函数可以处理图片、音频、视频等多媒体文件，支持多种格式转换和处理，例如对图片进行压缩、裁剪、打水印等操作。同时，云函数也提供了丰富的数据处理和分析功能，例如数据库操作、机器学习推理、文件存储等。

云函数的核心优势在于事件驱动、按需付费和弹性伸缩，用户只需编写最重要的“核心代码”，不再需要关心底层计算资源、服务器运维等操作。

2. web云函数

采用腾讯云 https://console.cloud.tencent.com/scf/list

新建一个web函数

直接在编辑器里写代码

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/get_name', methods=['POST'])
def get_name():
    try:
        # 获取请求体数据
        data = request.get_json()

        # 从请求体中获取 "name" 字段
        name = data.get('name')

        # 检查是否存在 "name" 字段
        if name is not None:
            return jsonify({'result': f'Hello, {name}!'})
        else:
            return jsonify({'error': 'Missing "name" field in the request body'}), 400

    except Exception as e:
        return jsonify({'error': str(e)}), 500


if __name__ == '__main__':
   app.run(host='0.0.0.0',port=9000)  # 必须是这个host port

3. 部署+发布新版本

上图的访问路径后面需要用到，鉴权方式没有设置（演示），正式的应用应该要设置

4. 测试

发一个上面请求路径的 web 请求，立即收到了处理结果

资源使用这里产生了一条记录

import requests

url = "https://service-*******apigw.com/release/get_name"

payload = {
    "name": "michael",
    "age": 18
}
headers = {"content-type": "application/json"}

response = requests.request("POST", url, json=payload, headers=headers)

print(response.text)
# {"result":"Hello, michael!"}

如果是一个计算时间很长的任务呢？

5. 事件函数

api 网关

配置事件函数网关触发

编写事件函数代码

# -*- coding: utf8 -*-
import json
import time
def main_handler(event, context):
    print("Received event: " + json.dumps(event, indent = 2)) 
    print("Received context: " + str(context))
    print("Hello world")
    body = json.loads(event['body'])
    #  time.sleep(60)
    return {
        "name": f"hello {body.get('name', 'no name')}",
        "event": json.dumps(event, indent = 2),
        "context": str(context),
    }

发送请求

{
	"name": "hello michael",
	"event": "{\n  \"body\": \"{\\r\\n    \\\"name\\\": \\\"michael\\\",\\r\\n    \\\"age\\\": 18\\r\\n}\",\n  \"headerParameters\": {},\n  \"headers\": {\n    \"accept\": \"*/*\",\n    \"accept-encoding\": \"gzip, deflate, br\",\n    \"cache-control\": \"no-cache\",\n    \"connection\": \"keep-alive\",\n    \"content-length\": \"43\",\n    \"content-type\": \"application/json\",\n    \"host\": \"service--13036992.bj.tencntapgw.com\",\n    \"requestsource\": \"APIGW\",\n    \"user-agent\": \"PostmanRuntime-ApipostRuntime/1.1.0\",\n    \"x-api-requestid\": \"cc2c773839019fecd1e3d4a1865a\",\n    \"x-api-scheme\": \"https\",\n    \"x-b3-traceid\": \"cc2c7738390ecd1ce03e3d4a1865a\",\n    \"x-qualifier\": \"$DEFAULT\"\n  },\n  \"httpMethod\": \"POST\",\n  \"isBase64Encoded\": false,\n  \"path\": \"/helloworld-event\",\n  \"pathParameters\": {},\n  \"queryString\": {},\n  \"queryStringParameters\": {},\n  \"requestContext\": {\n    \"httpMethod\": \"ANY\",\n    \"identity\": {},\n    \"path\": \"/helloworld-event\",\n    \"serviceId\": \"service-c9\",\n    \"sourceIp\": \"17.6.0.2\",\n    \"stage\": \"release\"\n  }\n}",
	"context": "{'memory_limit_in_mb': 128, 'time_limit_in_ms': 3000, 'request_id': 'cc2c7738390198888803e3d4a185a', 'environment': '{\"SCF_NAMESPACE\":\"default\"}', 'environ': 'SCF_NAMESPACE=default;SCF_NAMESPACE=default', 'function_version': '$LATEST', 'function_name': 'helloworld-event', 'namespace': 'default', 'tencentcloud_region': 'ap-ing', 'tencentcloud_appid': '1888892', 'tencentcloud_uin': '1881'}"
}

在代码中加入 sleep 60 s

再次请求

{"errorCode":-1,"errorMessage":"Invoking task timed out after 3 seconds",
"requestId":"497f4e0a94003c01e27775","statusCode":433}

原因：指定函数的最长运行时间，可选值范围为1秒- 86400秒（24小时，一天够了吧），默认3秒，改成最大的24小时。

现在开启异步：

查看云日志：

现在还要在代码里获取请求id，并查询请求任务是否结束

# pip install tencentcloud-sdk-python-scf

import json
from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.scf.v20180416 import scf_client, models

info = {
        "name": "michael ming",
        "age": 18
    }
event = {
    "body": json.dumps(info)
}

# 密钥可前往https://console.cloud.tencent.com/cam/capi 获取
cred = credential.Credential("***", "****")
# 实例化一个http选项，可选的，没有特殊需求可以跳过
httpProfile = HttpProfile()
httpProfile.endpoint = "scf.ap-beijing.tencentcloudapi.com"

# 实例化一个client选项，可选的，没有特殊需求可以跳过
clientProfile = ClientProfile()
clientProfile.httpProfile = httpProfile

# 实例化要请求产品的client对象,clientProfile是可选的
client = scf_client.ScfClient(cred, "ap-beijing", clientProfile)

# 实例化一个请求对象,每个接口都会对应一个request对象
req = models.InvokeRequest()
params = {
    "FunctionName": "helloworld-event-async",
    "InvocationType": "Event",
    "ClientContext": json.dumps(event)
}
req.from_json_string(json.dumps(params))

# 返回的resp是一个InvokeResponse的实例，与请求对象对应
resp = client.Invoke(req)
# 输出json格式的字符串回包
print(resp.to_json_string())

输出：可以获取到 RequestId

{"Result": {"Log": "", "RetMsg": "", "ErrMsg": "", "MemUsage": 0, "Duration": 0, "BillDuration": 0, 
		"FunctionRequestId": "aae06af6-ffa5-4290-b1bb-c3199fd14f03", "InvokeResult": 0}, 
"RequestId": "aae06af6-ffa5-4290-b1bb-c3199fd14f03"}

参考：https://cloud.tencent.com/document/product/583/51519

事件调用成功，返回信息只包含 RequestId
事件执行结果需要在函数代码逻辑中 自行实现 回调特定的 API 或者 发送通知消息（发kafka、写mysql、ES等）

获取异步执行的状态

req = models.GetAsyncEventStatusRequest()
params = {
    "InvokeRequestId": resp.RequestId,  #  resp = client.Invoke(req) 的 response
}
req.from_json_string(json.dumps(params))
resp = client.GetAsyncEventStatus(req)
print(resp)

输出

{"Result": {"Status": "FAILED", "StatusCode": 430, 
		"InvokeRequestId": "aae06af6-ffa5-4290-b1bb-c3199fd14f03"}, 
 "RequestId": "d3b78aee-20e7-409c-8f3b-081dff53331e"}

{"Result": {"Status": "FINISHED", "StatusCode": 200, 
		"InvokeRequestId": "20d41399-1f36-4a50-a8dd-2b38f056b605"}, 
 "RequestId": "2d6ae115-ff2b-4852-8150-3ff68f67d9a0"}