雲端原生已經從一個技術趨勢演變為現代軟體開發的標準範式。它不僅是將應用程式部署到雲端平台這麼簡單,而是一套完整的設計理念、架構模式、開發流程與運維實踐的集合。雲端原生應用程式天生為雲端環境設計,充分利用雲端平台的彈性、可擴展性、高可用性與按需付費的特性,透過微服務架構、容器化技術、動態編排、持續交付等手段,實現快速迭代、靈活部署與自動化運維。然而,從傳統的單體應用架構遷移到雲端原生架構並非易事,它涉及技術棧的轉變、團隊協作模式的調整、運維思維的革新,以及對分散式系統複雜性的深刻理解。本文將深入探討雲端原生應用程式開發的完整知識體系,從基礎概念到進階實踐,從架構設計到安全防護,為讀者提供建構現代化雲端原生應用的完整指南。無論您是正在考慮雲端遷移的架構師,還是希望提升技術能力的開發者,這份指南都將為您提供寶貴的參考與實戰經驗。
雲端原生架構設計原則與實踐
雲端原生的核心理念可以用 Cloud Native Computing Foundation (CNCF) 的定義來概括,那就是建構可在現代化動態環境(如公有雲、私有雲與混合雲)中運行的可擴展應用程式。這些應用程式採用容器、服務網格、微服務、不可變基礎設施與宣告式 API 等技術,透過自動化與編排實現鬆散耦合的系統,使工程師能夠輕鬆地進行高頻率且可預測的重大變更。
十二要素應用程式(The Twelve-Factor App)是雲端原生設計的重要指導原則。這套方法論由 Heroku 的工程師提出,總結了構建現代化 SaaS 應用的最佳實踐。第一要素是基準代碼,一份基準代碼透過版本控制進行追蹤,對應多份部署。第二要素是依賴管理,明確宣告並隔離依賴項。第三要素是配置管理,在環境中儲存配置而非代碼中。第四要素是後端服務,將後端服務視為附加資源。第五要素是建構、發布、運行,嚴格分離建構、發布與運行階段。第六要素是進程,以一個或多個無狀態進程運行應用。第七要素是連接埠綁定,透過連接埠綁定提供服務。第八要素是並發性,透過進程模型進行擴展。第九要素是易處理性,快速啟動與優雅終止。第十要素是開發與生產環境等價,盡可能保持開發、預發布與生產環境相同。第十一要素是日誌,將日誌視為事件流。第十二要素是管理進程,將管理任務作為一次性進程運行。
微服務架構是雲端原生的核心模式之一。與傳統的單體應用不同,微服務將應用程式拆分為多個小型、自治的服務,每個服務專注於單一業務能力,擁有獨立的資料存儲,透過輕量級的通訊機制協作。這種架構帶來了諸多優勢,包括獨立部署、技術棧自由、團隊自治、故障隔離等。然而,微服務也引入了分散式系統的複雜性,如服務間通訊、資料一致性、分散式追蹤、服務治理等挑戰。
from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity
import consul
import logging
import os
import socket
from datetime import timedelta
from functools import wraps
import requests
from circuitbreaker import circuit
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 微服務基礎類別
class MicroserviceBase:
"""
微服務基礎類別
提供服務註冊、健康檢查、配置管理等通用功能
"""
def __init__(self, app_name, app_port):
"""
初始化微服務
Parameters:
-----------
app_name : str
服務名稱
app_port : int
服務連接埠
"""
self.app = Flask(app_name)
self.app_name = app_name
self.app_port = app_port
self.service_id = f"{app_name}-{socket.gethostname()}-{app_port}"
# 資料庫配置
self.app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv(
'DATABASE_URL',
'postgresql://user:password@localhost:5432/mydb'
)
self.app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# JWT 配置
self.app.config['JWT_SECRET_KEY'] = os.getenv(
'JWT_SECRET_KEY',
'your-secret-key'
)
self.app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1)
# 初始化擴展
self.db = SQLAlchemy(self.app)
self.migrate = Migrate(self.app, self.db)
self.jwt = JWTManager(self.app)
# Consul 客戶端
self.consul_client = consul.Consul(
host=os.getenv('CONSUL_HOST', 'localhost'),
port=int(os.getenv('CONSUL_PORT', 8500))
)
# 註冊健康檢查端點
self.register_health_check()
logger.info(f"Microservice {self.app_name} initialized")
def register_health_check(self):
"""註冊健康檢查端點"""
@self.app.route('/health', methods=['GET'])
def health_check():
"""健康檢查端點"""
try:
# 檢查資料庫連線
self.db.session.execute('SELECT 1')
return jsonify({
'status': 'healthy',
'service': self.app_name,
'service_id': self.service_id
}), 200
except Exception as e:
logger.error(f"Health check failed: {e}")
return jsonify({
'status': 'unhealthy',
'error': str(e)
}), 503
def register_to_consul(self):
"""註冊服務到 Consul"""
try:
self.consul_client.agent.service.register(
name=self.app_name,
service_id=self.service_id,
address=socket.gethostbyname(socket.gethostname()),
port=self.app_port,
tags=['microservice', 'python', 'flask'],
check=consul.Check.http(
f"http://{socket.gethostbyname(socket.gethostname())}:{self.app_port}/health",
interval='10s',
timeout='5s',
deregister='30s'
)
)
logger.info(f"Service {self.service_id} registered to Consul")
except Exception as e:
logger.error(f"Failed to register to Consul: {e}")
def deregister_from_consul(self):
"""從 Consul 註銷服務"""
try:
self.consul_client.agent.service.deregister(self.service_id)
logger.info(f"Service {self.service_id} deregistered from Consul")
except Exception as e:
logger.error(f"Failed to deregister from Consul: {e}")
def discover_service(self, service_name):
"""
服務發現
Parameters:
-----------
service_name : str
要發現的服務名稱
Returns:
--------
str or None
服務 URL 或 None
"""
try:
_, services = self.consul_client.health.service(
service_name,
passing=True
)
if services:
service = services[0]
address = service['Service']['Address']
port = service['Service']['Port']
return f"http://{address}:{port}"
else:
logger.warning(f"No healthy instance found for service: {service_name}")
return None
except Exception as e:
logger.error(f"Service discovery failed: {e}")
return None
def run(self):
"""啟動微服務"""
try:
self.register_to_consul()
self.app.run(
host='0.0.0.0',
port=self.app_port,
debug=False
)
finally:
self.deregister_from_consul()
# 使用者服務範例
class UserService(MicroserviceBase):
"""使用者服務"""
def __init__(self):
super().__init__('user-service', 5001)
self.init_routes()
def init_routes(self):
"""初始化路由"""
@self.app.route('/api/users/register', methods=['POST'])
def register_user():
"""使用者註冊"""
try:
data = request.json
# 驗證必填欄位
if not all(k in data for k in ['username', 'email', 'password']):
return jsonify({'error': 'Missing required fields'}), 400
# 檢查使用者是否已存在
existing_user = User.query.filter(
(User.username == data['username']) |
(User.email == data['email'])
).first()
if existing_user:
return jsonify({'error': 'User already exists'}), 409
# 建立新使用者
user = User(
username=data['username'],
email=data['email']
)
user.set_password(data['password'])
self.db.session.add(user)
self.db.session.commit()
logger.info(f"User registered: {user.username}")
return jsonify({
'message': 'User registered successfully',
'user_id': user.id
}), 201
except Exception as e:
self.db.session.rollback()
logger.error(f"User registration failed: {e}")
return jsonify({'error': 'Internal server error'}), 500
@self.app.route('/api/users/login', methods=['POST'])
def login_user():
"""使用者登入"""
try:
data = request.json
if not all(k in data for k in ['username', 'password']):
return jsonify({'error': 'Missing credentials'}), 400
user = User.query.filter_by(username=data['username']).first()
if not user or not user.check_password(data['password']):
return jsonify({'error': 'Invalid credentials'}), 401
# 生成 JWT Token
access_token = create_access_token(identity=user.id)
logger.info(f"User logged in: {user.username}")
return jsonify({
'access_token': access_token,
'user_id': user.id,
'username': user.username
}), 200
except Exception as e:
logger.error(f"Login failed: {e}")
return jsonify({'error': 'Internal server error'}), 500
@self.app.route('/api/users/profile', methods=['GET'])
@jwt_required()
def get_profile():
"""取得使用者資料"""
try:
user_id = get_jwt_identity()
user = User.query.get(user_id)
if not user:
return jsonify({'error': 'User not found'}), 404
return jsonify({
'user_id': user.id,
'username': user.username,
'email': user.email
}), 200
except Exception as e:
logger.error(f"Get profile failed: {e}")
return jsonify({'error': 'Internal server error'}), 500
# 訂單服務範例(使用斷路器模式)
class OrderService(MicroserviceBase):
"""訂單服務"""
def __init__(self):
super().__init__('order-service', 5002)
self.init_routes()
@circuit(failure_threshold=5, recovery_timeout=60)
def call_user_service(self, user_id, token):
"""
呼叫使用者服務(帶斷路器保護)
Parameters:
-----------
user_id : int
使用者 ID
token : str
JWT Token
Returns:
--------
dict or None
使用者資料或 None
"""
user_service_url = self.discover_service('user-service')
if not user_service_url:
raise Exception("User service not available")
response = requests.get(
f"{user_service_url}/api/users/profile",
headers={'Authorization': f'Bearer {token}'},
timeout=5
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"User service returned {response.status_code}")
def init_routes(self):
"""初始化路由"""
@self.app.route('/api/orders', methods=['POST'])
@jwt_required()
def create_order():
"""建立訂單"""
try:
user_id = get_jwt_identity()
data = request.json
# 驗證使用者(透過服務間通訊)
token = request.headers.get('Authorization').split(' ')[1]
try:
user_data = self.call_user_service(user_id, token)
except Exception as e:
logger.error(f"Failed to verify user: {e}")
return jsonify({
'error': 'User verification failed',
'message': 'User service temporarily unavailable'
}), 503
# 建立訂單
order = Order(
user_id=user_id,
product_id=data['product_id'],
quantity=data['quantity'],
total_amount=data['total_amount']
)
self.db.session.add(order)
self.db.session.commit()
logger.info(f"Order created: {order.id} by user {user_id}")
return jsonify({
'message': 'Order created successfully',
'order_id': order.id
}), 201
except Exception as e:
self.db.session.rollback()
logger.error(f"Order creation failed: {e}")
return jsonify({'error': 'Internal server error'}), 500
# 資料模型
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime
class User:
"""使用者模型"""
__tablename__ = 'users'
id = None # 實際應使用 SQLAlchemy 定義
username = None
email = None
password_hash = None
created_at = None
def set_password(self, password):
"""設定密碼"""
self.password_hash = generate_password_hash(password)
def check_password(self, password):
"""驗證密碼"""
return check_password_hash(self.password_hash, password)
class Order:
"""訂單模型"""
__tablename__ = 'orders'
id = None
user_id = None
product_id = None
quantity = None
total_amount = None
status = None
created_at = None
# 啟動服務範例
if __name__ == "__main__":
import sys
if len(sys.argv) > 1 and sys.argv[1] == 'user':
service = UserService()
else:
service = OrderService()
service.run()
這個完整的微服務實作展示了雲端原生應用的核心特性。透過 Consul 實現服務註冊與發現,讓服務能夠動態地找到彼此。健康檢查機制確保只有健康的服務實例接收流量。斷路器模式保護系統免受級聯故障的影響。JWT 認證提供安全的身份驗證機制。這些都是建構可靠微服務系統的關鍵要素。
@startuml
!define PLANTUML_FORMAT svg
!theme _none_
skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 140
package "雲端原生應用架構" {
actor "使用者" as user
package "邊緣層" {
rectangle "API Gateway" as gateway
rectangle "負載平衡器" as lb
}
package "服務網格層" {
rectangle "Istio Proxy" as istio
rectangle "服務發現\nConsul" as consul
}
package "微服務層" {
rectangle "使用者服務" as user_svc
rectangle "訂單服務" as order_svc
rectangle "產品服務" as product_svc
rectangle "支付服務" as payment_svc
}
package "資料層" {
database "使用者資料庫" as user_db
database "訂單資料庫" as order_db
database "產品資料庫" as product_db
queue "訊息佇列\nRabbitMQ" as mq
database "快取層\nRedis" as cache
}
package "基礎設施層" {
rectangle "容器編排\nKubernetes" as k8s
rectangle "容器執行環境\nDocker" as docker
rectangle "監控與日誌\nPrometheus\nELK Stack" as monitor
}
}
user --> gateway
gateway --> lb
lb --> istio
istio --> consul
istio --> user_svc
istio --> order_svc
istio --> product_svc
istio --> payment_svc
user_svc --> user_db
order_svc --> order_db
product_svc --> product_db
order_svc --> mq
payment_svc --> mq
user_svc --> cache
product_svc --> cache
user_svc -[hidden]-> k8s
order_svc -[hidden]-> k8s
product_svc -[hidden]-> k8s
payment_svc -[hidden]-> k8s
k8s --> docker
k8s --> monitor
@enduml容器化技術與 Kubernetes 編排
容器化是雲端原生的基石技術。Docker 透過將應用程式及其依賴項封裝為獨立的容器映像,實現了"建構一次,到處運行"的理念。容器相比傳統虛擬機器更加輕量,啟動速度更快,資源利用率更高。然而,容器本身只是執行環境的封裝,要實現大規模容器的管理與編排,需要 Kubernetes 這樣的容器編排平台。
Dockerfile 是建構容器映像的藍圖。編寫高效的 Dockerfile 需要遵循多項最佳實踐。首先是選擇合適的基礎映像,盡量使用官方映像或輕量級的 Alpine Linux 映像以減少映像大小。其次是利用多階段建構,將建構環境與運行環境分離,避免將建構工具和中間檔案打包進最終映像。再者是合理組織指令順序,將變動頻率低的指令放在前面以充分利用層快取。最後是注意映像大小與安全性,定期更新基礎映像,移除不必要的檔案,使用非 root 使用者運行容器。
# 多階段建構範例 - Python 微服務
# 第一階段:建構環境
FROM python:3.11-slim as builder
# 設定工作目錄
WORKDIR /app
# 複製依賴檔案
COPY requirements.txt .
# 安裝依賴到本地目錄
RUN pip install --user --no-cache-dir --no-warn-script-location \
-r requirements.txt
# 第二階段:運行環境
FROM python:3.11-slim
# 建立非 root 使用者
RUN useradd -m -u 1000 appuser
# 設定工作目錄
WORKDIR /app
# 從建構階段複製依賴
COPY --from=builder /root/.local /home/appuser/.local
# 複製應用程式代碼
COPY --chown=appuser:appuser . .
# 設定環境變數
ENV PATH=/home/appuser/.local/bin:$PATH \
PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1
# 切換到非 root 使用者
USER appuser
# 暴露連接埠
EXPOSE 5000
# 健康檢查
HEALTHCHECK --interval=30s --timeout=3s --start-period=40s --retries=3 \
CMD python -c "import requests; requests.get('http://localhost:5000/health')"
# 啟動命令
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "4", "--timeout", "120", "app:app"]
Docker Compose 提供了簡單的多容器編排能力,適合開發環境與小規模部署。透過一個 YAML 檔案定義多個服務及其依賴關係,可以一鍵啟動整個應用堆疊。
version: '3.8'
services:
# PostgreSQL 資料庫
postgres:
image: postgres:15-alpine
container_name: postgres_db
environment:
POSTGRES_DB: ${DB_NAME:-myapp}
POSTGRES_USER: ${DB_USER:-postgres}
POSTGRES_PASSWORD: ${DB_PASSWORD:-postgres}
POSTGRES_INITDB_ARGS: "--encoding=UTF8 --locale=C"
volumes:
- postgres_data:/var/lib/postgresql/data
- ./init-scripts:/docker-entrypoint-initdb.d
ports:
- "5432:5432"
networks:
- app_network
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres"]
interval: 10s
timeout: 5s
retries: 5
restart: unless-stopped
# Redis 快取
redis:
image: redis:7-alpine
container_name: redis_cache
command: redis-server --appendonly yes --requirepass ${REDIS_PASSWORD:-redis}
volumes:
- redis_data:/data
ports:
- "6379:6379"
networks:
- app_network
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 3s
retries: 5
restart: unless-stopped
# RabbitMQ 訊息佇列
rabbitmq:
image: rabbitmq:3-management-alpine
container_name: rabbitmq_mq
environment:
RABBITMQ_DEFAULT_USER: ${RABBITMQ_USER:-admin}
RABBITMQ_DEFAULT_PASS: ${RABBITMQ_PASSWORD:-admin}
volumes:
- rabbitmq_data:/var/lib/rabbitmq
ports:
- "5672:5672"
- "15672:15672"
networks:
- app_network
healthcheck:
test: ["CMD", "rabbitmq-diagnostics", "ping"]
interval: 30s
timeout: 10s
retries: 5
restart: unless-stopped
# 使用者服務
user-service:
build:
context: ./services/user-service
dockerfile: Dockerfile
container_name: user_service
environment:
DATABASE_URL: postgresql://${DB_USER:-postgres}:${DB_PASSWORD:-postgres}@postgres:5432/users
REDIS_URL: redis://:${REDIS_PASSWORD:-redis}@redis:6379/0
JWT_SECRET_KEY: ${JWT_SECRET_KEY:-your-secret-key}
CONSUL_HOST: consul
CONSUL_PORT: 8500
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
consul:
condition: service_healthy
ports:
- "5001:5000"
networks:
- app_network
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
interval: 30s
timeout: 5s
retries: 3
start_period: 40s
restart: unless-stopped
deploy:
resources:
limits:
cpus: '0.5'
memory: 512M
reservations:
cpus: '0.25'
memory: 256M
# 訂單服務
order-service:
build:
context: ./services/order-service
dockerfile: Dockerfile
container_name: order_service
environment:
DATABASE_URL: postgresql://${DB_USER:-postgres}:${DB_PASSWORD:-postgres}@postgres:5432/orders
REDIS_URL: redis://:${REDIS_PASSWORD:-redis}@redis:6379/1
RABBITMQ_URL: amqp://${RABBITMQ_USER:-admin}:${RABBITMQ_PASSWORD:-admin}@rabbitmq:5672/
JWT_SECRET_KEY: ${JWT_SECRET_KEY:-your-secret-key}
CONSUL_HOST: consul
CONSUL_PORT: 8500
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
rabbitmq:
condition: service_healthy
consul:
condition: service_healthy
ports:
- "5002:5000"
networks:
- app_network
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
interval: 30s
timeout: 5s
retries: 3
start_period: 40s
restart: unless-stopped
deploy:
resources:
limits:
cpus: '0.5'
memory: 512M
# Consul 服務發現
consul:
image: consul:latest
container_name: consul_server
command: agent -server -ui -bootstrap-expect=1 -client=0.0.0.0
ports:
- "8500:8500"
- "8600:8600/udp"
networks:
- app_network
healthcheck:
test: ["CMD", "consul", "members"]
interval: 10s
timeout: 5s
retries: 5
restart: unless-stopped
# Nginx API Gateway
nginx:
image: nginx:alpine
container_name: api_gateway
volumes:
- ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
- ./nginx/ssl:/etc/nginx/ssl:ro
ports:
- "80:80"
- "443:443"
networks:
- app_network
depends_on:
- user-service
- order-service
healthcheck:
test: ["CMD", "wget", "--quiet", "--tries=1", "--spider", "http://localhost/health"]
interval: 30s
timeout: 5s
retries: 3
restart: unless-stopped
volumes:
postgres_data:
driver: local
redis_data:
driver: local
rabbitmq_data:
driver: local
networks:
app_network:
driver: bridge
ipam:
config:
- subnet: 172.28.0.0/16
Kubernetes 是生產環境中的容器編排標準。它提供了服務發現與負載平衡、儲存編排、自動化部署與回滾、自動裝箱、自我修復、密鑰與配置管理等強大功能。
# Kubernetes 部署範例 - 使用者服務
---
# Deployment - 管理 Pod 的建立與更新
apiVersion: apps/v1
kind: Deployment
metadata:
name: user-service
namespace: production
labels:
app: user-service
version: v1.0.0
tier: backend
spec:
replicas: 3
revisionHistoryLimit: 10
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 1
maxUnavailable: 0
selector:
matchLabels:
app: user-service
template:
metadata:
labels:
app: user-service
version: v1.0.0
spec:
# 服務帳戶
serviceAccountName: user-service-sa
# 容器配置
containers:
- name: user-service
image: myregistry.azurecr.io/user-service:v1.0.0
imagePullPolicy: IfNotPresent
ports:
- containerPort: 5000
name: http
protocol: TCP
# 環境變數
env:
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: database-credentials
key: connection-string
- name: REDIS_URL
valueFrom:
configMapKeyRef:
name: redis-config
key: redis-url
- name: JWT_SECRET_KEY
valueFrom:
secretKeyRef:
name: jwt-secret
key: secret-key
# 資源限制
resources:
requests:
cpu: 250m
memory: 256Mi
limits:
cpu: 500m
memory: 512Mi
# 健康檢查
livenessProbe:
httpGet:
path: /health
port: 5000
initialDelaySeconds: 30
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 3
readinessProbe:
httpGet:
path: /health
port: 5000
initialDelaySeconds: 10
periodSeconds: 5
timeoutSeconds: 3
failureThreshold: 3
# 掛載配置與密鑰
volumeMounts:
- name: config-volume
mountPath: /app/config
readOnly: true
# Volume 定義
volumes:
- name: config-volume
configMap:
name: user-service-config
# 親和性規則
affinity:
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchExpressions:
- key: app
operator: In
values:
- user-service
topologyKey: kubernetes.io/hostname
---
# Service - 服務發現與負載平衡
apiVersion: v1
kind: Service
metadata:
name: user-service
namespace: production
labels:
app: user-service
spec:
type: ClusterIP
selector:
app: user-service
ports:
- port: 80
targetPort: 5000
protocol: TCP
name: http
sessionAffinity: None
---
# HorizontalPodAutoscaler - 自動擴展
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: user-service-hpa
namespace: production
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: user-service
minReplicas: 3
maxReplicas: 10
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
behavior:
scaleDown:
stabilizationWindowSeconds: 300
policies:
- type: Percent
value: 50
periodSeconds: 60
scaleUp:
stabilizationWindowSeconds: 0
policies:
- type: Percent
value: 100
periodSeconds: 15
- type: Pods
value: 2
periodSeconds: 15
selectPolicy: Max
---
# ConfigMap - 配置管理
apiVersion: v1
kind: ConfigMap
metadata:
name: user-service-config
namespace: production
data:
app.conf: |
LOG_LEVEL=INFO
MAX_CONNECTIONS=100
REQUEST_TIMEOUT=30
redis-config: |
REDIS_URL=redis://redis-service:6379/0
Serverless 無伺服器運算架構
Serverless 無伺服器運算代表了雲端原生的進一步演進。它讓開發者完全專注於業務邏輯,而將基礎設施管理完全交給雲端平台。無伺服器不是真的沒有伺服器,而是伺服器對開發者完全透明,由平台自動管理擴展、高可用、安全修補等運維工作。
AWS Lambda 是最流行的 Serverless 計算服務。Lambda 函式採用事件驅動模型,可以被各種事件源觸發,如 API Gateway 請求、S3 檔案上傳、DynamoDB 資料變更、CloudWatch 定時任務等。Lambda 按實際執行時間計費,空閒時不產生費用,非常適合間歇性、突發性的工作負載。
import json
import boto3
import os
import logging
from datetime import datetime
from decimal import Decimal
import uuid
# 設定日誌
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# AWS 服務客戶端
dynamodb = boto3.resource('dynamodb')
s3 = boto3.client('s3')
sns = boto3.client('sns')
sqs = boto3.client('sqs')
# 環境變數
TABLE_NAME = os.environ['DYNAMODB_TABLE']
BUCKET_NAME = os.environ['S3_BUCKET']
SNS_TOPIC_ARN = os.environ['SNS_TOPIC_ARN']
class DecimalEncoder(json.JSONEncoder):
"""DynamoDB Decimal 型別的 JSON 編碼器"""
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return super(DecimalEncoder, self).default(obj)
def lambda_handler(event, context):
"""
Lambda 主處理函式
處理來自 API Gateway 的請求
Parameters:
-----------
event : dict
事件資料
context : LambdaContext
執行上下文
Returns:
--------
dict
HTTP 回應
"""
try:
# 記錄請求資訊
logger.info(f"Received event: {json.dumps(event)}")
# 解析 HTTP 方法與路徑
http_method = event['httpMethod']
path = event['path']
# 路由處理
if http_method == 'POST' and path == '/api/orders':
return create_order(event, context)
elif http_method == 'GET' and path == '/api/orders':
return list_orders(event, context)
elif http_method == 'GET' and path.startswith('/api/orders/'):
return get_order(event, context)
elif http_method == 'PUT' and path.startswith('/api/orders/'):
return update_order(event, context)
else:
return {
'statusCode': 404,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*'
},
'body': json.dumps({'error': 'Not Found'})
}
except Exception as e:
logger.error(f"Error processing request: {str(e)}", exc_info=True)
return {
'statusCode': 500,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*'
},
'body': json.dumps({
'error': 'Internal Server Error',
'message': str(e)
})
}
def create_order(event, context):
"""
建立訂單
Parameters:
-----------
event : dict
事件資料
context : LambdaContext
執行上下文
Returns:
--------
dict
HTTP 回應
"""
try:
# 解析請求資料
body = json.loads(event['body'])
# 驗證必填欄位
required_fields = ['user_id', 'product_id', 'quantity', 'price']
for field in required_fields:
if field not in body:
return {
'statusCode': 400,
'headers': {'Content-Type': 'application/json'},
'body': json.dumps({'error': f'Missing field: {field}'})
}
# 建立訂單資料
order_id = str(uuid.uuid4())
timestamp = datetime.utcnow().isoformat()
order = {
'order_id': order_id,
'user_id': body['user_id'],
'product_id': body['product_id'],
'quantity': Decimal(str(body['quantity'])),
'price': Decimal(str(body['price'])),
'total_amount': Decimal(str(body['quantity'])) * Decimal(str(body['price'])),
'status': 'pending',
'created_at': timestamp,
'updated_at': timestamp
}
# 儲存到 DynamoDB
table = dynamodb.Table(TABLE_NAME)
table.put_item(Item=order)
logger.info(f"Order created: {order_id}")
# 發送 SNS 通知
send_notification(
'Order Created',
f"New order {order_id} has been created"
)
# 將訂單資料儲存到 S3(用於資料湖)
save_to_s3(order_id, order)
return {
'statusCode': 201,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*'
},
'body': json.dumps({
'message': 'Order created successfully',
'order_id': order_id
}, cls=DecimalEncoder)
}
except Exception as e:
logger.error(f"Error creating order: {str(e)}", exc_info=True)
return {
'statusCode': 500,
'headers': {'Content-Type': 'application/json'},
'body': json.dumps({'error': 'Failed to create order'})
}
def list_orders(event, context):
"""
列出訂單
支援分頁與篩選
Parameters:
-----------
event : dict
事件資料
context : LambdaContext
執行上下文
Returns:
--------
dict
HTTP 回應
"""
try:
# 解析查詢參數
params = event.get('queryStringParameters', {}) or {}
user_id = params.get('user_id')
limit = int(params.get('limit', 10))
last_evaluated_key = params.get('last_key')
table = dynamodb.Table(TABLE_NAME)
# 建構查詢參數
scan_kwargs = {
'Limit': min(limit, 100) # 最多 100 筆
}
if last_evaluated_key:
scan_kwargs['ExclusiveStartKey'] = json.loads(last_evaluated_key)
if user_id:
scan_kwargs['FilterExpression'] = 'user_id = :user_id'
scan_kwargs['ExpressionAttributeValues'] = {':user_id': user_id}
# 執行查詢
response = table.scan(**scan_kwargs)
# 準備回應
result = {
'orders': response['Items'],
'count': len(response['Items'])
}
if 'LastEvaluatedKey' in response:
result['last_key'] = json.dumps(response['LastEvaluatedKey'])
return {
'statusCode': 200,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*'
},
'body': json.dumps(result, cls=DecimalEncoder)
}
except Exception as e:
logger.error(f"Error listing orders: {str(e)}", exc_info=True)
return {
'statusCode': 500,
'headers': {'Content-Type': 'application/json'},
'body': json.dumps({'error': 'Failed to list orders'})
}
def get_order(event, context):
"""
取得訂單詳情
Parameters:
-----------
event : dict
事件資料
context : LambdaContext
執行上下文
Returns:
--------
dict
HTTP 回應
"""
try:
# 從路徑中提取訂單 ID
order_id = event['pathParameters']['order_id']
table = dynamodb.Table(TABLE_NAME)
response = table.get_item(Key={'order_id': order_id})
if 'Item' not in response:
return {
'statusCode': 404,
'headers': {'Content-Type': 'application/json'},
'body': json.dumps({'error': 'Order not found'})
}
return {
'statusCode': 200,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*'
},
'body': json.dumps(response['Item'], cls=DecimalEncoder)
}
except Exception as e:
logger.error(f"Error getting order: {str(e)}", exc_info=True)
return {
'statusCode': 500,
'headers': {'Content-Type': 'application/json'},
'body': json.dumps({'error': 'Failed to get order'})
}
def update_order(event, context):
"""
更新訂單狀態
Parameters:
-----------
event : dict
事件資料
context : LambdaContext
執行上下文
Returns:
--------
dict
HTTP 回應
"""
try:
order_id = event['pathParameters']['order_id']
body = json.loads(event['body'])
# 只允許更新狀態
new_status = body.get('status')
if not new_status:
return {
'statusCode': 400,
'headers': {'Content-Type': 'application/json'},
'body': json.dumps({'error': 'Status is required'})
}
table = dynamodb.Table(TABLE_NAME)
# 更新訂單
response = table.update_item(
Key={'order_id': order_id},
UpdateExpression='SET #status = :status, updated_at = :updated_at',
ExpressionAttributeNames={'#status': 'status'},
ExpressionAttributeValues={
':status': new_status,
':updated_at': datetime.utcnow().isoformat()
},
ReturnValues='ALL_NEW'
)
logger.info(f"Order updated: {order_id}")
# 發送狀態變更通知
send_notification(
'Order Status Changed',
f"Order {order_id} status changed to {new_status}"
)
return {
'statusCode': 200,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*'
},
'body': json.dumps(response['Attributes'], cls=DecimalEncoder)
}
except Exception as e:
logger.error(f"Error updating order: {str(e)}", exc_info=True)
return {
'statusCode': 500,
'headers': {'Content-Type': 'application/json'},
'body': json.dumps({'error': 'Failed to update order'})
}
def send_notification(subject, message):
"""發送 SNS 通知"""
try:
sns.publish(
TopicArn=SNS_TOPIC_ARN,
Subject=subject,
Message=message
)
logger.info(f"Notification sent: {subject}")
except Exception as e:
logger.error(f"Failed to send notification: {str(e)}")
def save_to_s3(order_id, order_data):
"""將訂單資料儲存到 S3"""
try:
key = f"orders/{datetime.utcnow().strftime('%Y/%m/%d')}/{order_id}.json"
s3.put_object(
Bucket=BUCKET_NAME,
Key=key,
Body=json.dumps(order_data, cls=DecimalEncoder),
ContentType='application/json'
)
logger.info(f"Order saved to S3: {key}")
except Exception as e:
logger.error(f"Failed to save to S3: {str(e)}")
這個完整的 Lambda 函式實作展示了 Serverless 應用的典型模式。透過與 DynamoDB、S3、SNS 等 AWS 服務的整合,實現了完整的訂單管理功能。函式採用事件驅動架構,支援 RESTful API,並包含完善的錯誤處理與日誌記錄。
玄貓認為,雲端原生代表了軟體架構的根本性轉變,它不僅是技術的升級,更是思維方式的革新。從單體應用到微服務,從虛擬機器到容器,從固定容量到彈性擴展,從手動運維到自動化管理,每一步都在改變我們構建與運行應用程式的方式。然而,雲端原生也帶來了新的挑戰,分散式系統的複雜性、服務間通訊的可靠性、資料一致性的保證、監控與除錯的困難,都需要團隊具備更高的技術能力與成熟的工程實踐。成功的雲端原生轉型不僅需要技術的投入,更需要組織文化的變革、團隊能力的提升與持續學習的文化。隨著雲端技術的持續演進,Serverless、邊緣運算、服務網格、可觀測性等新技術將持續湧現,掌握雲端原生的核心原則與最佳實踐,才能在快速變化的技術浪潮中保持競爭力,建構真正靈活、可靠且高效的現代化應用系統。
