Serverless实战：用函数计算实现图片压缩服务

一、项目概述

本文将指导你使用阿里云函数计算（Function Compute）构建一个高性能、低成本的图片压缩服务。该服务支持：

• 自动压缩上传的图片
• 保持图片质量的同时显著减小文件大小
• 支持常见图片格式（JPEG, PNG, WebP）
• 通过HTTP API触发

二、架构设计

用户上传图片 → API网关 → 函数计算 → 图片处理 → 结果存储到OSS → 返回压缩后URL

三、环境准备

1. 开通服务

• 阿里云函数计算
• 对象存储OSS（用于存储原始和压缩后图片）
• API网关（可选，用于提供HTTP接口）

2. 安装工具

# 安装Serverless Devs工具
npm install @serverless-devs/s -g

# 配置阿里云凭证
s config add --AccessKeyID xxx --AccessKeySecret yyy --AccountID zzz

四、代码实现

1. 项目结构

image-compression/
├── src/
│   ├── index.py           # 主函数入口
│   ├── processor.py       # 图片处理逻辑
│   └── requirements.txt   # Python依赖
├── s.yaml                 # Serverless配置
└── template.yml           # 资源模板

2. 核心代码实现

src/processor.py - 图片处理模块

import io
from PIL import Image
import tinify

class ImageProcessor:
    def __init__(self, tinify_key=None):
        if tinify_key:
            tinify.key = tinify_key
    
    def compress_with_pillow(self, image_data, quality=85, format='JPEG'):
        """使用Pillow进行基础压缩"""
        img = Image.open(io.BytesIO(image_data))
        
        # 转换模式为RGB（如果是RGBA）
        if img.mode in ('RGBA', 'LA'):
            background = Image.new('RGB', img.size, (255, 255, 255))
            if img.mode == 'LA':
                background.paste(img, mask=img.split()[1])
            else:
                background.paste(img, mask=img.split()[3])
            img = background
        
        # 压缩图片
        output = io.BytesIO()
        img.save(output, format=format, quality=quality, optimize=True)
        
        return output.getvalue()
    
    def compress_with_tinify(self, image_data):
        """使用Tinify API进行高级压缩（需要API Key）"""
        try:
            compressed = tinify.from_buffer(image_data).to_buffer()
            return compressed
        except Exception as e:
            raise Exception(f"Tinify压缩失败: {str(e)}")
    
    def get_image_info(self, image_data):
        """获取图片信息"""
        img = Image.open(io.BytesIO(image_data))
        return {
            'format': img.format,
            'size': img.size,
            'mode': img.mode,
            'original_size': len(image_data)
        }

src/index.py - 函数计算入口

import json
import base64
import os
import time
from processor import ImageProcessor
import oss2

def handler(event, context):
    """
    处理图片压缩请求
    """
    # 解析请求
    try:
        request = json.loads(event.decode('utf-8') if isinstance(event, bytes) else event)
    except:
        request = event
    
    # 从环境变量获取配置
    oss_endpoint = os.environ.get('OSS_ENDPOINT')
    oss_bucket = os.environ.get('OSS_BUCKET')
    oss_access_key = os.environ.get('OSS_ACCESS_KEY_ID')
    oss_secret_key = os.environ.get('OSS_ACCESS_KEY_SECRET')
    tinify_key = os.environ.get('TINIFY_API_KEY', None)
    
    # 初始化OSS客户端
    auth = oss2.Auth(oss_access_key, oss_secret_key)
    bucket = oss2.Bucket(auth, oss_endpoint, oss_bucket)
    
    # 获取图片数据
    image_data = get_image_data(request, bucket)
    
    # 创建处理器
    processor = ImageProcessor(tinify_key)
    
    # 获取原始图片信息
    original_info = processor.get_image_info(image_data)
    
    # 选择压缩方式
    if tinify_key:
        compressed_data = processor.compress_with_tinify(image_data)
    else:
        # 根据图片格式选择压缩参数
        if original_info['format'] == 'PNG':
            compressed_data = processor.compress_with_pillow(image_data, quality=90, format='PNG')
        else:
            compressed_data = processor.compress_with_pillow(image_data, quality=85, format='JPEG')
    
    # 保存压缩后的图片
    compressed_info = {
        'format': original_info['format'],
        'size': len(compressed_data),
        'compression_ratio': round((1 - len(compressed_data) / original_info['original_size']) * 100, 2)
    }
    
    # 生成文件名
    timestamp = int(time.time())
    compressed_key = f"compressed/{timestamp}_{original_info['format'].lower()}.{original_info['format'].lower()}"
    
    # 上传到OSS
    bucket.put_object(compressed_key, compressed_data)
    
    # 生成访问URL
    url = bucket.sign_url('GET', compressed_key, 3600)
    
    return {
        'success': True,
        'data': {
            'original': original_info,
            'compressed': compressed_info,
            'url': url,
            'key': compressed_key
        }
    }

def get_image_data(request, bucket):
    """从不同来源获取图片数据"""
    # 方式1：直接传递base64编码的图片
    if 'image_base64' in request:
        return base64.b64decode(request['image_base64'])
    
    # 方式2：传递OSS路径
    elif 'oss_key' in request:
        obj = bucket.get_object(request['oss_key'])
        return obj.read()
    
    # 方式3：通过HTTP上传的图片
    elif 'body' in request and request.get('isBase64Encoded', False):
        return base64.b64decode(request['body'])
    
    else:
        raise ValueError("无法获取图片数据，请提供image_base64或oss_key参数")

3. 依赖文件

src/requirements.txt

Pillow==9.5.0
tinify==1.6.1
oss2==2.18.1

五、Serverless配置

s.yaml

edition: 3.0.0
name: image-compression
access: "default"

vars:
  region: "cn-hangzhou"
  service:
    name: "image-service"
    description: "图片处理服务"

resources:
  image-compression:
    component: fc3
    actions:
      pre-deploy:
        - run: pip install -r src/requirements.txt -t .
          path: ./src
    props:
      region: ${vars.region}
      service: ${vars.service}
      function:
        name: "compress-image"
        description: "图片压缩函数"
        runtime: python3.10
        codeUri: ./src
        handler: index.handler
        memorySize: 512
        timeout: 30
        environmentVariables:
          OSS_ENDPOINT: ${env.OSS_ENDPOINT}
          OSS_BUCKET: ${env.OSS_BUCKET}
          OSS_ACCESS_KEY_ID: ${env.OSS_ACCESS_KEY_ID}
          OSS_ACCESS_KEY_SECRET: ${env.OSS_ACCESS_KEY_SECRET}
          TINIFY_API_KEY: ${env.TINIFY_API_KEY}
      triggers:
        - name: http-trigger
          type: http
          config:
            authType: anonymous
            methods:
              - POST

六、部署与测试

1. 部署服务

# 进入项目目录
cd image-compression

# 部署到阿里云
s deploy --use-local

2. 环境变量配置

创建.env文件：

OSS_ENDPOINT=oss-cn-hangzhou.aliyuncs.com
OSS_BUCKET=your-bucket-name
OSS_ACCESS_KEY_ID=your-access-key
OSS_ACCESS_KEY_SECRET=your-secret-key
TINIFY_API_KEY=your-tinify-key # 可选

3. 测试函数

Python测试脚本：

import requests
import base64
import json

# 测试URL（部署后获取）
url = "https://your-service-url"

# 准备测试图片
with open("test.jpg", "rb") as f:
    image_base64 = base64.b64encode(f.read()).decode('utf-8')

# 请求数据
payload = {
    "image_base64": image_base64,
    "options": {
        "quality": 85,
        "format": "JPEG"
    }
}

# 发送请求
response = requests.post(url, json=payload)
result = response.json()

print(f"压缩前: {result['data']['original']['original_size']} bytes")
print(f"压缩后: {result['data']['compressed']['size']} bytes")
print(f"压缩率: {result['data']['compressed']['compression_ratio']}%")
print(f"下载链接: {result['data']['url']}")

七、进阶功能

1. 支持WebP格式

在processor.py中添加：

def convert_to_webp(self, image_data, quality=80):
    """转换为WebP格式"""
    img = Image.open(io.BytesIO(image_data))
    output = io.BytesIO()
    
    # 如果图片有透明度，保持透明度
    if img.mode in ('RGBA', 'LA'):
        img.save(output, 'WEBP', quality=quality, lossless=False)
    else:
        img = img.convert('RGB')
        img.save(output, 'WEBP', quality=quality)
    
    return output.getvalue()

2. 批量处理支持

def batch_compress(event, context):
    """批量压缩处理"""
    requests = event.get('requests', [])
    results = []
    
    for req in requests:
        try:
            result = handler(req, context)
            results.append(result)
        except Exception as e:
            results.append({
                'success': False,
                'error': str(e)
            })
    
    return {
        'success': True,
        'results': results
    }

3. 添加图片水印

def add_watermark(self, image_data, watermark_text):
    """添加文字水印"""
    img = Image.open(io.BytesIO(image_data)).convert('RGBA')
    
    # 创建水印图层
    txt = Image.new('RGBA', img.size, (255,255,255,0))
    
    # 绘制水印
    from PIL import ImageDraw, ImageFont
    draw = ImageDraw.Draw(txt)
    
    # 使用默认字体或自定义字体
    try:
        font = ImageFont.truetype("arial.ttf", 40)
    except:
        font = ImageFont.load_default()
    
    # 计算水印位置
    text_width, text_height = draw.textsize(watermark_text, font)
    x = img.width - text_width - 10
    y = img.height - text_height - 10
    
    # 添加水印
    draw.text((x, y), watermark_text, font=font, fill=(255,255,255,128))
    
    # 合并图层
    watermarked = Image.alpha_composite(img, txt)
    
    # 保存结果
    output = io.BytesIO()
    watermarked.save(output, format=img.format or 'PNG')
    
    return output.getvalue()

八、性能优化建议

1. 冷启动优化：

   • 增加内存配置（512MB以上）
   • 使用预留实例
   • 精简依赖包大小

2. 成本控制：

   • 设置函数超时时间（建议30秒）
   • 使用异步调用处理大图片
   • 配置合理的重试策略

3. 监控与告警：

   # 在s.yaml中添加监控配置
   customDomains:
     - domainName: image.your-domain.com
       protocol: HTTP
       routeConfigs:
         - path: /*
           functionName: compress-image
   

九、总结

通过本文的指导，你已经成功构建了一个基于Serverless的图片压缩服务。该方案具有以下优势：

✅ 优点

• 低成本：按实际使用量计费，无请求时不产生费用
• 弹性伸缩：自动处理高并发请求
• 免运维：无需管理服务器
• 高可用：函数计算提供多可用区容灾

⚠️ 注意事项

1. 函数计算有最大运行时间限制（最长15分钟）
2. 单个函数包大小有限制（建议不超过50MB）
3. OSS存储费用需单独考虑

🚀 扩展方向

1. 添加图片格式转换功能
2. 集成CDN加速访问
3. 实现图片智能裁剪（人脸识别）
4. 添加API限流和鉴权

这个方案特别适合以下场景：

• 移动应用图片上传处理
• 电商网站商品图片优化
• 内容管理系统中的图片处理
• 批量图片处理任务

你可以根据实际需求调整压缩参数，平衡图片质量和文件大小，实现最佳用户体验。