无服务器架构Serverless入门

# serverless_cost_calc.py —— Serverless 月度成本粗算
# 单价为占位示例值，实际请以 2026 年各云官网实时报价为准
PRICE_PER_REQUEST = 0.0000013      # 元 / 次
PRICE_PER_GB_SECOND = 0.00012000   # 元 / GB·秒
GB_SECOND_FREE = 40000             # 每月免费 GB·秒示例

def monthly_cost(calls_per_month, avg_ms, memory_mb):
    gb = memory_mb / 1024
    gb_second = calls_per_month * gb * (avg_ms / 1000)
    billable = max(gb_second - GB_SECOND_FREE, 0)
    request_fee = calls_per_month * PRICE_PER_REQUEST
    resource_fee = billable * PRICE_PER_GB_SECOND
    # 打印两个候选内存配置，便于对比择优
    return {
        "requests": round(request_fee, 2),
        "gb_second_total": round(gb_second, 2),
        "resource": round(resource_fee, 2),
        "total": round(request_fee + resource_fee, 2),
    }

if __name__ == "__main__":
    for mem in (512, 1024, 2048):
        # 内存翻倍通常执行时间缩短，这里用经验值近似
        avg_ms = {512: 420, 1024: 210, 2048: 120}[mem]
        r = monthly_cost(3_000_000, avg_ms, mem)
        print(f"内存 {mem:>5} MB | 平均 {avg_ms:>4} ms | "
              f"GB·秒 {r['gb_second_total']:>10} | 月度合计 {r['total']:>8} 元")

# 1) 精简依赖包：只打包运行时真正需要的部分（Python 示例）
pip install --target ./package -r requirements.txt --no-compile --no-cache-dir

# 2) 剔除测试目录、__pycache__、.git，减小代码包体积
find ./package -type d -name "__pycache__" -prune -exec rm -rf {} +
find ./package -type d -name "tests" -prune -exec rm -rf {} +

# 3) 进入 package 目录打包，避免出现多余的一层路径
cd package && zip -r -q ../function.zip . -x "*.pyc" "*.dist-info/*" && cd ..
zip -q -g function.zip handler.py

# 4) 查看最终体积，建议控制在 50 MB 以内（解压后常控制在 250 MB 内）
ls -lh function.zip

# handler.py —— Python 3.12 函数入口示例
import json
import time


def _response(status: int, body: dict) -> dict:
    """统一构造 HTTP 响应结构，便于网关识别"""
    return {
        "statusCode": status,
        "headers": {"Content-Type": "application/json; charset=utf-8"},
        "body": json.dumps(body, ensure_ascii=False),
    }


def main_handler(event, context):
    # event 中包含 httpMethod、body、queryString 等字段
    try:
        payload = json.loads(event.get("body") or "{}")
    except json.JSONDecodeError:
        return _response(400, {"ok": False, "error": "invalid JSON body"})

    name = payload.get("name", "").strip()
    if not name:
        return _response(422, {"ok": False, "error": "field 'name' is required"})

    # 业务逻辑：这里只做一次轻量处理，重活应丢给消息队列异步执行
    started = time.time()
    result = {"message": f"hello, {name}", "length": len(name)}
    cost_ms = round((time.time() - started) * 1000, 2)

    return _response(200, {"ok": True, "cost_ms": cost_ms, "data": result})

# serverless.yml —— 函数、触发器与网关配置
service: serverless-demo

provider:
  name: cloud-vendor          # 替换为你的云厂商标识
  runtime: python3.12
  region: ap-guangzhou
  memorySize: 1024            # MB，需与 CPU 配额联动调整
  timeout: 30                 # 秒，上限以厂商文档为准
  environment:
    LOG_LEVEL: INFO

functions:
  hello:
    handler: handler.main_handler
    description: "Serverless 入门示例函数"
    events:
      # HTTP 触发器：对外暴露 REST 接口
      - http:
          path: /hello
          method: POST
      # 定时触发器：每天 02:00 执行一次（Cron 五段式）
      - timer:
          cron: "0 0 2 * * *"
          enable: true

plugins:
  - serverless-python-packaging

# 本地安装依赖并跑一次单元自检
python -m pip install -r requirements.txt
python -c "
import handler
evt = {'body': '{\"name\": \"ops\"}'}
print(handler.main_handler(evt, None))
"

# 通过框架 CLI 部署（以 serverless framework 为例）
npx serverless deploy --stage prod

# 部署后立即验证线上返回码与耗时
curl -s -o /dev/null -w "http_code=%{http_code} time=%{time_total}s\n" \
  -X POST https://example.com/hello \
  -H "Content-Type: application/json" \
  -d '{"name":"ops"}'