# 亿级流量架构演进：从单机部署到 K8s 编排的 8 个关键阶段

面对从零到亿级流量的业务增长，系统架构需要经历一系列关键的演进阶段。本文将详细解析高并发架构从单机部署到 Kubernetes 编排的完整演进路径。

## 阶段一：单体应用与基础部署

### 初始单体架构

初创阶段采用简单的单体架构，快速验证业务模式：

```python

# app.py - Flask 单体应用示例

from flask import Flask, request, jsonify

import sqlite3

import threading

app = Flask(__name__)

# 简单的数据库连接

def get_db_connection():

conn = sqlite3.connect('app.db')

conn.row_factory = sqlite3.Row

return conn

# 用户注册接口

@app.route('/register', methods=['POST'])

def register():

data = request.json

conn = get_db_connection()

cursor = conn.cursor()

try:

cursor.execute(

'INSERT INTO users (username, email) VALUES (?, ?)',

(data['username'], data['email'])

)

conn.commit()

return jsonify({'status': 'success'})

except Exception as e:

return jsonify({'error': str(e)}), 400

# 产品列表接口

@app.route('/products')

def get_products():

conn = get_db_connection()

products = conn.execute('SELECT * FROM products').fetchall()

return jsonify([dict(product) for product in products])

if __name__ == '__main__':

app.run(host='0.0.0.0', port=5000, debug=True)

```

基础部署脚本：

```bash

#!/bin/bash

# deploy.sh - 单机部署脚本

echo "开始单机部署..."

# 安装依赖

pip install -r requirements.txt

# 初始化数据库

python init_db.py

# 启动应用

nohup python app.py > app.log 2>&1 &

echo "应用启动完成，监听端口 5000"

```

## 阶段二：应用与数据分离

### 数据库独立部署

将数据库从应用中分离，使用专业数据库服务：

```yaml

# docker-compose.yml - 应用与数据库分离

version: '3.8'

services:

app:

build: .

ports:

- "5000:5000"

environment:

- DATABASE_URL=postgresql://user:pass@db:5432/app_db

depends_on:

- db

db:

image: postgres:13

environment:

POSTGRES_DB: app_db

POSTGRES_USER: user

POSTGRES_PASSWORD: pass

volumes:

- db_data:/var/lib/postgresql/data

ports:

- "5432:5432"

volumes:

db_data:

```

数据库连接优化：

```python

# database.py - 数据库连接池

import psycopg2

<"jntaikang.cn">

<"btyhx.com">

<"yapuz.com">

from psycopg2 import pool

class DatabaseManager:

def __init__(self):

self.connection_pool = psycopg2.pool.SimpleConnectionPool(

1, 20, # 最小1，最大20连接

user="user",

password="pass",

host="db",

port="5432",

database="app_db"

)

def get_connection(self):

return self.connection_pool.getconn()

def return_connection(self, conn):

self.connection_pool.putconn(conn)

# 使用连接池

db_manager = DatabaseManager()

@app.route('/users')

def get_users():

conn = db_manager.get_connection()

try:

cursor = conn.cursor()

cursor.execute('SELECT * FROM users LIMIT 100')

users = cursor.fetchall()

return jsonify(users)

finally:

db_manager.return_connection(conn)

```

## 阶段三：引入缓存层

### Redis 缓存集成

添加 Redis 缓存减轻数据库压力：

```python

# cache.py - Redis 缓存层

import redis

import json

import pickle

class CacheManager:

def __init__(self):

self.redis_client = redis.Redis(

host='redis-cache',

port=6379,

db=0,

decode_responses=True

)

def get_cached_data(self, key):

"""获取缓存数据"""

cached = self.redis_client.get(key)

if cached:

return json.loads(cached)

return None

def set_cached_data(self, key, data, expire=3600):

"""设置缓存数据"""

self.redis_client.setex(

key,

expire,

json.dumps(data, default=str)

)

def invalidate_pattern(self, pattern):

"""按模式清除缓存"""

keys = self.redis_client.keys(pattern)

if keys:

self.redis_client.delete(*keys)

# 使用缓存的业务逻辑

cache_manager = CacheManager()

@app.route('/products')

<"hbgufen.com">

<"okjiabao.com">

def get_products():

cache_key = "all_products"

# 先尝试从缓存获取

cached_products = cache_manager.get_cached_data(cache_key)

if cached_products:

return jsonify(cached_products)

# 缓存未命中，查询数据库

conn = get_db_connection()

products = conn.execute('SELECT * FROM products').fetchall()

product_list = [dict(product) for product in products]

# 写入缓存

cache_manager.set_cached_data(cache_key, product_list, expire=1800)

return jsonify(product_list)

```

缓存架构配置：

```yaml

# docker-compose with cache

version: '3.8'

services:

app:

build: .

environment:

- REDIS_URL=redis://redis-cache:6379

depends_on:

- db

- redis-cache

redis-cache:

image: redis:6-alpine

command: redis-server --appendonly yes

volumes:

- redis_data:/data

volumes:

redis_data:

```

## 阶段四：水平扩展与负载均衡

### Nginx 负载均衡配置

通过负载均衡实现应用层水平扩展：

```nginx

# nginx.conf - 负载均衡配置

upstream app_servers {

server app1:5000 weight=3;

server app2:5000 weight=2;

server app3:5000 weight=2;

# 会话保持

sticky cookie srv_id expires=1h domain=.example.com path=/;

}

server {

listen 80;

server_name api.example.com;

# 健康检查

location /health {

proxy_pass http://app_servers;

proxy_next_upstream error timeout invalid_header http_500 http_502 http_503;

}

location / {

proxy_pass http://app_servers;

proxy_set_header Host $host;

proxy_set_header X-Real-IP $remote_addr;

proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

# 超时配置

proxy_connect_timeout 30s;

proxy_send_timeout 30s;

proxy_read_timeout 30s;

}

# 静态文件缓存

location ~* \.(js|css|png|jpg|jpeg|gif|ico)$ {

expires 1y;

add_header Cache-Control "public, immutable";

}

```

多应用实例部署：

```yaml

# docker-compose.scale.yml

version: '3.8'

services:

nginx:

image: nginx:alpine

ports:

- "80:80"

volumes:

- ./nginx.conf:/etc/nginx/nginx.conf

depends_on:

- app1

- app2

- app3

app1:

build: .

environment:

- INSTANCE_ID=app1

deploy:

replicas: 1

app2:

build: .

environment:

- INSTANCE_ID=app2

deploy:

replicas: 1

app3:

build: .

environment:

- INSTANCE_ID=app3

deploy:

replicas: 1

```

## 阶段五：数据库读写分离

### 主从复制与读写分离

实现数据库的读写分离：

```python

# database_router.py - 数据库路由

class DatabaseRouter:

<"kuww.com.cn">

<"zhongguoinfo.com">

def __init__(self):

self.write_pool = psycopg2.pool.SimpleConnectionPool(

1, 10,

host='db-master',

database='app_db'

)

self.read_pool = psycopg2.pool.SimpleConnectionPool(

1, 20,

host='db-replica',

database='app_db'

)

def get_write_connection(self):

return self.write_pool.getconn()

def get_read_connection(self):

return self.read_pool.getconn()

def route_connection(self, operation_type='read'):

"""根据操作类型路由到不同数据库"""

if operation_type in ['write', 'insert', 'update', 'delete']:

return self.get_write_connection()

else:

return self.get_read_connection()

# 在业务代码中使用

db_router = DatabaseRouter()

@app.route('/users', methods=['GET'])

def get_users():

# 读操作使用从库

conn = db_router.route_connection('read')

# ... 查询逻辑

@app.route('/users', methods=['POST'])

def create_user():

# 写操作使用主库

conn = db_router.route_connection('write')

# ... 插入逻辑

```

数据库主从配置：

```yaml

# docker-compose.db-cluster.yml

services:

db-master:

image: postgres:13

environment:

- POSTGRES_REPLICATION_MODE=master

- POSTGRES_REPLICATION_USER=replicator

- POSTGRES_REPLICATION_PASSWORD=replica_pass

volumes:

- master_data:/var/lib/postgresql/data

db-replica:

image: postgres:13

environment:

- POSTGRES_REPLICATION_MODE=slave

- POSTGRES_REPLICATION_USER=replicator

- POSTGRES_REPLICATION_PASSWORD=replica_pass

- POSTGRES_MASTER_SERVICE_NAME=db-master

depends_on:

- db-master

volumes:

- replica_data:/var/lib/postgresql/data

```

## 阶段六：微服务架构转型

### 服务拆分与 API 网关

将单体应用拆分为微服务：

```python

# user_service/app.py - 用户服务

from flask import Flask, jsonify

import os

app = Flask(__name__)

@app.route('/users/<user_id>')

def get_user(user_id):

# 用户服务业务逻辑

return jsonify({

'id': user_id,

'name': '用户示例',

'service': 'user-service'

})

if __name__ == '__main__':

app.run(host='0.0.0.0', port=5001)

# product_service/app.py - 商品服务

@app.route('/products/<product_id>')

def get_product(product_id):

# 商品服务业务逻辑

return jsonify({

'id': product_id,

'name': '商品示例',

'service': 'product-service'

})

```

API 网关配置：

```yaml

# api-gateway/config.yml

routes:

- path: /users/**

service: user-service

strip_prefix: false

- path: /products/**

service: product-service

strip_prefix: false

- path: /orders/**

service: order-service

strip_prefix: false

services:

user-service:

urls:

- http://user-service:5001

product-service:

urls:

- http://product-service:5002

order-service:

urls:

- http://order-service:5003

```

## 阶段七：容器化与编排

### Docker 化与 Kubernetes 部署

完整的 Kubernetes 部署配置：

```yaml

# k8s/deployment.yml

apiVersion: apps/v1

kind: Deployment

metadata:

labels:

app: user-service

spec:

replicas: 3

selector:

matchLabels:

app: user-service

template:

metadata:

labels:

app: user-service

spec:

containers:

- name: user-service

image: myregistry/user-service:v1.2.3

ports:

- containerPort: 5001

env:

- name: DATABASE_URL

valueFrom:

secretKeyRef:

key: connection-string

resources:

requests:

memory: "128Mi"

cpu: "100m"

limits:

memory: "256Mi"

cpu: "200m"

livenessProbe:

httpGet:

path: /health

port: 5001

initialDelaySeconds: 30

periodSeconds: 10

readinessProbe:

httpGet:

path: /health

port: 5001

initialDelaySeconds: 5

periodSeconds: 5

---

apiVersion: v1

kind: Service

metadata:

spec:

selector:

app: user-service

ports:

- port: 80

targetPort: 5001

type: ClusterIP

```

服务网格配置：

```yaml

# k8s/istio-gateway.yml

apiVersion: networking.istio.io/v1alpha3

kind: Gateway

metadata:

spec:

selector:

istio: ingressgateway

servers:

- port:

number: 80

protocol: HTTP

hosts:

- "api.example.com"

---

apiVersion: networking.istio.io/v1alpha3

kind: VirtualService

metadata:

spec:

hosts:

- "api.example.com"

gateways:

- api-gateway

http:

- match:

- uri:

prefix: "/users"

route:

- destination:

host: user-service

port:

number: 80

- match:

- uri:

prefix: "/products"

route:

- destination:

host: product-service

port:

number: 80

```

## 阶段八：全链路监控与自动化

### 可观测性栈部署

完整的监控和日志系统：

```yaml

# k8s/monitoring.yml

apiVersion: v1

kind: ConfigMap

metadata:

data:

prometheus.yml: |

global:

scrape_interval: 15s

evaluation_interval: 15s

scrape_configs:

- job_name: 'user-service'

static_configs:

- targets: ['user-service:5001']

metrics_path: /metrics

- job_name: 'product-service'

static_configs:

- targets: ['product-service:5002']

metrics_path: /metrics

- job_name: 'kubernetes-nodes'

kubernetes_sd_configs:

- role: node

---

apiVersion: apps/v1

kind: Deployment

metadata:

spec:

replicas: 1

template:

spec:

containers:

- name: prometheus

image: prom/prometheus:v2.30.3

ports:

- containerPort: 9090

volumeMounts:

- name: config-volume

mountPath: /etc/prometheus

resources:

requests:

memory: "512Mi"

cpu: "300m"

volumes:

- name: config-volume

configMap:

```

应用性能监控集成：

```python

# monitoring.py - 应用监控集成

from prometheus_client import Counter, Histogram, generate_latest

import time

from flask import Response

# 定义指标

REQUEST_COUNT = Counter(

'http_requests_total',

'Total HTTP Requests',

['method', 'endpoint', 'status']

)

REQUEST_DURATION = Histogram(

'http_request_duration_seconds',

'HTTP request duration in seconds',

['method', 'endpoint']

)

@app.route('/metrics')

def metrics():

return Response(generate_latest(), mimetype='text/plain')

@app.before_request

def before_request():

request.start_time = time.time()

@app.after_request

def after_request(response):

# 记录请求指标

REQUEST_COUNT.labels(

method=request.method,

endpoint=request.path,

status=response.status_code

).inc()

# 记录请求耗时

if hasattr(request, 'start_time'):

REQUEST_DURATION.labels(

method=request.method,

endpoint=request.path

).observe(time.time() - request.start_time)

return response

```

### CI/CD 流水线

完整的自动化部署流水线：

```yaml

# .github/workflows/deploy.yml

on:

push:

branches: [ main ]

jobs:

test:

runs-on: ubuntu-latest

steps:

- uses: actions/checkout@v2

- name: Run Tests

run: |

docker-compose -f docker-compose.test.yml up -d

pytest tests/ --cov=app

- name: Security Scan

uses: aquasecurity/trivy-action@master

with:

image-ref: 'myapp:latest'

build-and-deploy:

needs: test

runs-on: ubuntu-latest

steps:

- name: Build and Push Docker Image

run: |

docker build -t myregistry/myapp:${{ github.sha }} .

docker push myregistry/myapp:${{ github.sha }}

- name: Deploy to Kubernetes

run: |

kubectl set image deployment/myapp \

myapp=myregistry/myapp:${{ github.sha }}

kubectl rollout status deployment/myapp

- name: Run Integration Tests

run: |

./scripts/run-integration-tests.sh

```

通过这 8 个关键阶段的演进，系统架构从简单的单机部署逐步发展为能够支撑亿级流量的分布式系统。每个阶段都针对特定的瓶颈和挑战，通过技术升级和架构优化，确保系统在业务增长过程中始终保持高可用性和可扩展性。

ITPub博客

亿级流量架构演进：从单机部署到 K8s 编排的 8 个关键阶段

最新文章