【系统架构设计百科】无状态设计：扩展的第一步也是最难的一步

<!-- pom.xml -->
<dependency>
    <groupId>org.springframework.session</groupId>
    <artifactId>spring-session-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

# application.yml
spring:
  session:
    store-type: redis
    timeout: 1800
  redis:
    host: redis-cluster.internal
    port: 6379
    password: ${REDIS_PASSWORD}
    lettuce:
      pool:
        max-active: 50
        max-idle: 10
        min-idle: 5

// app.js
const session = require('express-session');
const RedisStore = require('connect-redis').default;
const { createClient } = require('redis');

const redisClient = createClient({
  url: 'redis://redis-cluster.internal:6379',
  password: process.env.REDIS_PASSWORD,
});
redisClient.connect();

app.use(session({
  store: new RedisStore({ client: redisClient }),
  secret: process.env.SESSION_SECRET,
  resave: false,
  saveUninitialized: false,
  cookie: {
    secure: true,
    httpOnly: true,
    maxAge: 1800000, // 30 分钟
  },
}));

// 方案二示例：本地缓存 + Redis 失效广播
@Component
public class ProductCacheManager {

    private final Cache<String, Product> localCache = Caffeine.newBuilder()
        .maximumSize(10000)
        .expireAfterWrite(Duration.ofMinutes(5))
        .build();

    @Autowired
    private RedisTemplate<String, Product> redisTemplate;

    @Autowired
    private RedisMessageListenerContainer listenerContainer;

    @PostConstruct
    public void init() {
        // 监听缓存失效通知
        listenerContainer.addMessageListener(
            (message, pattern) -> {
                String productId = new String(message.getBody());
                localCache.invalidate(productId);
            },
            new ChannelTopic("cache:product:invalidate")
        );
    }

    public Product getProduct(String productId) {
        // 先查本地缓存
        Product product = localCache.getIfPresent(productId);
        if (product != null) {
            return product;
        }
        // 再查 Redis
        product = redisTemplate.opsForValue().get("product:" + productId);
        if (product != null) {
            localCache.put(productId, product);
            return product;
        }
        // 最后查数据库
        product = productRepository.findById(productId);
        if (product != null) {
            redisTemplate.opsForValue().set("product:" + productId, product,
                Duration.ofMinutes(30));
            localCache.put(productId, product);
        }
        return product;
    }

    public void updateProduct(Product product) {
        productRepository.save(product);
        redisTemplate.opsForValue().set("product:" + product.getId(), product,
            Duration.ofMinutes(30));
        // 广播失效通知，所有实例的本地缓存都会失效
        redisTemplate.convertAndSend("cache:product:invalidate", product.getId());
    }
}

# task_manager.py
import redis
import json
import time

class TaskManager:
    def __init__(self):
        self.redis = redis.Redis(host='redis-cluster.internal', port=6379)

    def create_task(self, task_id, total_steps):
        task_state = {
            'task_id': task_id,
            'status': 'running',
            'current_step': 0,
            'total_steps': total_steps,
            'created_at': time.time(),
            'updated_at': time.time(),
        }
        self.redis.set(f'task:{task_id}', json.dumps(task_state))
        self.redis.expire(f'task:{task_id}', 86400)  # 24 小时过期

    def update_progress(self, task_id, current_step):
        raw = self.redis.get(f'task:{task_id}')
        if not raw:
            raise ValueError(f'Task {task_id} not found')
        task_state = json.loads(raw)
        task_state['current_step'] = current_step
        task_state['updated_at'] = time.time()
        if current_step >= task_state['total_steps']:
            task_state['status'] = 'completed'
        self.redis.set(f'task:{task_id}', json.dumps(task_state))

    def resume_task(self, task_id):
        """实例重启后，从上次的进度继续"""
        raw = self.redis.get(f'task:{task_id}')
        if not raw:
            return None
        state = json.loads(raw)
        return state['current_step'] if state['status'] == 'running' else None

// auth.js
const jwt = require('jsonwebtoken');
const fs = require('fs');

// 使用 RSA 非对称密钥
const privateKey = fs.readFileSync('/secrets/jwt-private.pem');
const publicKey = fs.readFileSync('/secrets/jwt-public.pem');

function issueToken(user) {
  const payload = {
    sub: user.id,
    role: user.role,
    permissions: user.permissions,
  };
  return jwt.sign(payload, privateKey, {
    algorithm: 'RS256',
    expiresIn: '15m',    // Access Token 短有效期
    issuer: 'auth.example.com',
  });
}

function issueRefreshToken(user) {
  return jwt.sign(
    { sub: user.id, type: 'refresh' },
    privateKey,
    { algorithm: 'RS256', expiresIn: '7d' }
  );
}

function verifyToken(token) {
  try {
    return jwt.verify(token, publicKey, {
      algorithms: ['RS256'],
      issuer: 'auth.example.com',
    });
  } catch (err) {
    if (err.name === 'TokenExpiredError') {
      throw new Error('Token expired');
    }
    throw new Error('Invalid token');
  }
}

// JwtAuthenticationFilter.java — 以下代码经删减，仅保留关键路径
@Component
public class JwtAuthenticationFilter extends OncePerRequestFilter {

    private final RSAPublicKey publicKey;

    @Override
    protected void doFilterInternal(HttpServletRequest request,
                                     HttpServletResponse response,
                                     FilterChain chain)
            throws ServletException, IOException {

        String header = request.getHeader("Authorization");
        if (header == null || !header.startsWith("Bearer ")) {
            chain.doFilter(request, response);
            return;
        }

        try {
            SignedJWT signedJWT = SignedJWT.parse(header.substring(7));
            if (!signedJWT.verify(new RSASSAVerifier(publicKey))) {
                response.setStatus(401);
                return;
            }
            JWTClaimsSet claims = signedJWT.getJWTClaimsSet();
            if (claims.getExpirationTime().before(new Date())) {
                response.setStatus(401);
                return;
            }
            SecurityContext ctx = SecurityContextHolder.createEmptyContext();
            ctx.setAuthentication(new JwtAuthentication(
                claims.getSubject(), claims.getStringClaim("role")));
            SecurityContextHolder.setContext(ctx);
        } catch (Exception e) {
            response.setStatus(401);
            return;
        }
        chain.doFilter(request, response);
    }
}

# jwt_blacklist.py
import redis
import time

class JWTBlacklist:
    def __init__(self):
        self.redis = redis.Redis(host='redis-cluster.internal', port=6379)

    def revoke(self, jti, exp):
        """将 JWT 加入黑名单，过期后自动清除"""
        ttl = exp - int(time.time())
        if ttl > 0:
            self.redis.setex(f'jwt:blacklist:{jti}', ttl, '1')

    def is_revoked(self, jti):
        return self.redis.exists(f'jwt:blacklist:{jti}') > 0

# 混合认证架构配置示例
# gateway-config.yml
routes:
  - path: /api/v1/**
    auth:
      type: jwt
      public-key-url: https://auth.internal/.well-known/jwks.json
      required-claims:
        - sub
        - role

  - path: /admin/**
    auth:
      type: session
      session-store: redis
      session-timeout: 1800
      require-mfa: true

  - path: /internal/**
    auth:
      type: mtls
      ca-cert: /certs/internal-ca.pem

# upload_service.py
import boto3
import uuid
from datetime import datetime

class UploadService:
    def __init__(self):
        self.s3 = boto3.client('s3',
            endpoint_url='https://s3.example.com',
            aws_access_key_id='...',
            aws_secret_access_key='...',
        )
        self.bucket = 'user-uploads'

    def generate_presigned_url(self, user_id, filename, content_type, file_size):
        # 校验文件类型和大小
        allowed_types = ['image/jpeg', 'image/png', 'application/pdf']
        if content_type not in allowed_types:
            raise ValueError(f'Unsupported content type: {content_type}')
        if file_size > 100 * 1024 * 1024:  # 100MB 上限
            raise ValueError('File too large')

        # 生成唯一的存储路径
        file_key = f'uploads/{user_id}/{datetime.now():%Y/%m}/{uuid.uuid4()}/{filename}'

        # 生成 Pre-signed URL，有效期 15 分钟
        presigned_url = self.s3.generate_presigned_url(
            'put_object',
            Params={
                'Bucket': self.bucket,
                'Key': file_key,
                'ContentType': content_type,
                'ContentLength': file_size,
            },
            ExpiresIn=900,
        )

        return {
            'upload_url': presigned_url,
            'file_key': file_key,
            'expires_in': 900,
        }

    def confirm_upload(self, file_key):
        """客户端上传完成后，验证文件确实存在"""
        try:
            response = self.s3.head_object(Bucket=self.bucket, Key=file_key)
            return {
                'file_key': file_key,
                'size': response['ContentLength'],
                'content_type': response['ContentType'],
                'etag': response['ETag'],
            }
        except self.s3.exceptions.ClientError:
            raise ValueError('File not found')

// upload.js
async function uploadFile(file) {
  const { upload_url, file_key } = await fetch('/api/upload/presign', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      filename: file.name, content_type: file.type, file_size: file.size,
    }),
  }).then(r => r.json());

  await fetch(upload_url, {
    method: 'PUT',
    headers: { 'Content-Type': file.type },
    body: file,
  });

  await fetch('/api/upload/confirm', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ file_key }),
  });
}

// upload_proxy.go
package main

import (
    "context"
    "fmt"
    "net/http"

    "github.com/aws/aws-sdk-go-v2/service/s3"
)

func streamUploadHandler(s3Client *s3.Client) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        contentType := r.Header.Get("Content-Type")
        contentLength := r.ContentLength
        fileKey := fmt.Sprintf("uploads/%s", r.URL.Query().Get("key"))

        // 直接把 request body 流式传到 S3，不在本地缓存
        _, err := s3Client.PutObject(context.Background(), &s3.PutObjectInput{
            Bucket:        stringPtr("user-uploads"),
            Key:           stringPtr(fileKey),
            Body:          r.Body, // 流式传输，不缓存到内存或磁盘
            ContentType:   stringPtr(contentType),
            ContentLength: &contentLength,
        })
        if err != nil {
            http.Error(w, "Upload failed", http.StatusInternalServerError)
            return
        }

        w.WriteHeader(http.StatusCreated)
    }
}

func stringPtr(s string) *string { return &s }

// ws-server.js
const WebSocket = require('ws');
const Redis = require('ioredis');

const INSTANCE_ID = process.env.INSTANCE_ID || 'instance-1';
const redisPub = new Redis('redis://redis-cluster.internal:6379');
const redisSub = new Redis('redis://redis-cluster.internal:6379');

// 本地连接注册表：userId -> WebSocket
const localConnections = new Map();

const wss = new WebSocket.Server({ port: 8080 });

wss.on('connection', (ws, req) => {
  const userId = authenticateFromRequest(req);
  if (!userId) {
    ws.close(4001, 'Unauthorized');
    return;
  }

  // 注册到本地连接表
  localConnections.set(userId, ws);

  // 在 Redis 中注册连接位置
  redisPub.hset('ws:connections', userId, INSTANCE_ID);

  ws.on('message', async (data) => {
    const message = JSON.parse(data);
    if (message.type === 'direct') {
      await sendToUser(message.targetUserId, message.payload);
    } else if (message.type === 'broadcast') {
      await broadcastToChannel(message.channel, message.payload);
    }
  });

  ws.on('close', () => {
    localConnections.delete(userId);
    redisPub.hdel('ws:connections', userId);
  });
});

async function sendToUser(targetUserId, payload) {
  const localWs = localConnections.get(targetUserId);
  if (localWs && localWs.readyState === WebSocket.OPEN) {
    localWs.send(JSON.stringify(payload));
    return;
  }
  // 不在本地，通过 Redis Pub/Sub 转发到目标实例
  const targetInstance = await redisPub.hget('ws:connections', targetUserId);
  if (targetInstance) {
    redisPub.publish(`ws:instance:${targetInstance}`,
      JSON.stringify({ targetUserId, payload }));
  }
}

// 监听发给本实例的定向消息
redisSub.subscribe(`ws:instance:${INSTANCE_ID}`);
redisSub.on('message', (channel, message) => {
  const { targetUserId, payload } = JSON.parse(message);
  const localWs = localConnections.get(targetUserId);
  if (localWs && localWs.readyState === WebSocket.OPEN) {
    localWs.send(JSON.stringify(payload));
  }
});

// graceful-drain.js
const DRAIN_TIMEOUT = 30000; // 30 秒排空超时
let isDraining = false;

process.on('SIGTERM', () => {
  isDraining = true;

  // 通知所有已连接的客户端迁移
  for (const [userId, ws] of localConnections) {
    ws.send(JSON.stringify({
      type: 'system',
      action: 'reconnect',
      message: 'Server is shutting down, please reconnect',
    }));
  }

  // 等待客户端断开或超时强制关闭
  const drainTimer = setTimeout(() => {
    for (const [userId, ws] of localConnections) {
      ws.close(1001, 'Server shutting down');
    }
    process.exit(0);
  }, DRAIN_TIMEOUT);

  const checkInterval = setInterval(() => {
    if (localConnections.size === 0) {
      clearTimeout(drainTimer);
      clearInterval(checkInterval);
      process.exit(0);
    }
  }, 1000);
});

// 健康检查端点：排空期间返回 503，负载均衡器不再分配新连接
app.get('/health', (req, res) => {
  res.status(isDraining ? 503 : 200).json({
    status: isDraining ? 'draining' : 'healthy',
  });
});

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ws-server
spec:
  replicas: 3
  template:
    spec:
      terminationGracePeriodSeconds: 60  # 给足排空时间
      containers:
        - name: ws-server
          image: ws-server:latest
          ports:
            - containerPort: 8080
          readinessProbe:
            httpGet:
              path: /health
              port: 8080
            periodSeconds: 5
          lifecycle:
            preStop:
              exec:
                command: ["sleep", "5"]  # 等待负载均衡器感知

// SessionConfig.java
@Configuration
@EnableRedisHttpSession(maxInactiveIntervalInSeconds = 1800)
public class SessionConfig {

    @Bean
    public LettuceConnectionFactory connectionFactory() {
        RedisStandaloneConfiguration config = new RedisStandaloneConfiguration();
        config.setHostName("redis-cluster.internal");
        config.setPort(6379);
        config.setPassword("...");
        return new LettuceConnectionFactory(config);
    }

    @Bean
    public CookieSerializer cookieSerializer() {
        DefaultCookieSerializer serializer = new DefaultCookieSerializer();
        serializer.setCookieName("ORDER_SID");
        serializer.setUseHttpOnlyCookie(true);
        serializer.setUseSecureCookie(true);
        serializer.setSameSite("Lax");
        return serializer;
    }
}

// ProductCacheService.java — 以下代码经删减，仅保留关键路径
@Service
public class ProductCacheService {

    private final Cache<Long, ProductDTO> l1Cache = Caffeine.newBuilder()
        .maximumSize(5000)
        .expireAfterWrite(Duration.ofSeconds(30))
        .build();

    @Autowired private StringRedisTemplate redisTemplate;
    @Autowired private ProductMapper productMapper;

    public ProductDTO getProduct(Long productId) {
        // L1 本地缓存 -> L2 Redis -> L3 数据库，逐级回源并回填上层
        ProductDTO product = l1Cache.getIfPresent(productId);
        if (product != null) return product;

        String json = redisTemplate.opsForValue().get("product:" + productId);
        if (json != null) {
            product = JSON.parseObject(json, ProductDTO.class);
            l1Cache.put(productId, product);
            return product;
        }

        product = productMapper.selectById(productId);
        if (product != null) {
            redisTemplate.opsForValue().set("product:" + productId,
                JSON.toJSONString(product), Duration.ofMinutes(10));
            l1Cache.put(productId, product);
        }
        return product;
    }
}

// InvoiceService.java — 以下代码经删减，仅保留关键路径
@Service
public class InvoiceService {

    @Autowired private AmazonS3 s3Client;
    private static final String BUCKET = "order-invoices";

    public String generateAndUploadInvoice(Order order) {
        byte[] pdfBytes = pdfGenerator.generate(order);
        String key = String.format("invoices/%s/%s.pdf",
            order.getOrderDate().format(DateTimeFormatter.ofPattern("yyyy/MM")),
            order.getOrderId());
        ObjectMetadata meta = new ObjectMetadata();
        meta.setContentLength(pdfBytes.length);
        meta.setContentType("application/pdf");
        s3Client.putObject(new PutObjectRequest(BUCKET, key,
            new ByteArrayInputStream(pdfBytes), meta));
        return key;
    }

    public String getDownloadUrl(String fileKey) {
        Date expiration = new Date(System.currentTimeMillis() + 600_000);
        return s3Client.generatePresignedUrl(BUCKET, fileKey,
            expiration, HttpMethod.GET).toString();
    }
}

#!/bin/bash
# migrate_uploads.sh - 将三台实例上的历史文件迁移到 S3
for host in 10.0.1.1 10.0.1.2 10.0.1.3; do
    echo "Migrating files from ${host}..."
    ssh "${host}" "find /data/uploads -type f" | while read filepath; do
        relative_path="${filepath#/data/uploads/}"
        aws s3 cp "ssh://${host}${filepath}" \
            "s3://order-invoices/migrated/${relative_path}" --quiet
    done
done
echo "Migration complete."
aws s3 ls s3://order-invoices/migrated/ --recursive --summarize | tail -2

// 反面示例：Session 中存放自定义对象，字段变更后旧 Session 反序列化失败
session.setAttribute("user", new UserContext(userId, role, permissions));

// 正面示例：Session 中只存基础类型
session.setAttribute("userId", userId);
session.setAttribute("role", role);
session.setAttribute("permissions", String.join(",", permissions));