# Python MessagePack 服务
FROM python:3.12-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 暴露端口
EXPOSE 8080

# 启动服务
CMD ["python", "server.py"]

# requirements.txt
flask==3.0.0
msgpack==1.0.7
gunicorn==21.2.0
redis==5.0.1

# Go MessagePack 服务 - 多阶段构建
FROM golang:1.22-alpine AS builder

WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download

COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -o server .

# 最终镜像
FROM alpine:3.19

RUN apk --no-cache add ca-certificates
WORKDIR /app
COPY --from=builder /app/server .

EXPOSE 8080
CMD ["./server"]

// go.mod
module msgpack-service

go 1.22

require (
    github.com/vmihailenco/msgpack/v5 v5.4.1
    github.com/gofiber/fiber/v2 v2.52.0
)

# Node.js MessagePack 服务
FROM node:20-alpine

WORKDIR /app

COPY package*.json ./
RUN npm ci --only=production

COPY . .

EXPOSE 8080
CMD ["node", "server.js"]

{
  "dependencies": {
    "@msgpack/msgpack": "^3.0.0",
    "express": "^4.18.2"
  }
}

version: "3.9"

services:
  # ========== API 网关 (Go) ==========
  gateway:
    build: ./gateway
    ports:
      - "8080:8080"
    environment:
      - USER_SERVICE_URL=user-service:8081
      - ORDER_SERVICE_URL=order-service:8082
    depends_on:
      - user-service
      - order-service
    networks:
      - microservices

  # ========== 用户服务 (Python) ==========
  user-service:
    build: ./user-service
    expose:
      - "8081"
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
    networks:
      - microservices
    deploy:
      replicas: 2

  # ========== 订单服务 (Node.js) ==========
  order-service:
    build: ./order-service
    expose:
      - "8082"
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
    networks:
      - microservices

  # ========== Redis 缓存 ==========
  redis:
    image: redis:7-alpine
    expose:
      - "6379"
    volumes:
      - redis-data:/data
    networks:
      - microservices

volumes:
  redis-data:

networks:
  microservices:
    driver: bridge

# MessagePack RPC 协议

## 消息格式

所有消息使用长度前缀 + MessagePack 编码：

{
  "result": {"id": 123, "name": "Alice"},
  "error": null,
  "id": "req-uuid-001"
}

// order-service/server.js
import express from "express";
import { encode, decode } from "@msgpack/msgpack";

const app = express();
app.use(express.raw({ type: "application/x-msgpack" }));

// ========== 数据存储 ==========
const orders = [
  { id: 1001, userId: 1, items: ["商品A", "商品B"], total: 199.99, status: "paid" },
  { id: 1002, userId: 2, items: ["商品C"], total: 59.99, status: "pending" },
  { id: 1003, userId: 1, items: ["商品D", "商品E", "商品F"], total: 349.99, status: "shipped" },
];

// ========== MessagePack 中间件 ==========
function msgpackHandler(handler) {
  return async (req, res) => {
    try {
      let params = {};
      if (req.headers["content-type"] === "application/x-msgpack") {
        params = decode(req.body);
      } else if (req.query) {
        params = req.query;
      }
      
      const result = await handler(params, req);
      
      const encoded = encode(result);
      res.set("Content-Type", "application/x-msgpack");
      res.send(Buffer.from(encoded));
    } catch (err) {
      const error = encode({ error: err.message });
      res.set("Content-Type", "application/x-msgpack");
      res.status(500).send(Buffer.from(error));
    }
  };
}

// ========== 路由 ==========
app.get("/orders", msgpackHandler(async (params) => {
  const userId = parseInt(params.userId);
  const userOrders = userId
    ? orders.filter(o => o.userId === userId)
    : orders;
  return { result: userOrders, count: userOrders.length };
}));

app.get("/orders/:id", msgpackHandler(async (params, req) => {
  const id = parseInt(req.params.id);
  const order = orders.find(o => o.id === id);
  if (!order) return { error: "订单不存在" };
  return { result: order };
}));

// ========== 启动 ==========
app.listen(8082, () => console.log("订单服务启动在 8082"));

// gateway/main.go
package main

import (
    "bytes"
    "encoding/binary"
    "fmt"
    "io"
    "net/http"
    "os"
    "time"

    "github.com/gofiber/fiber/v2"
    "github.com/vmihailenco/msgpack/v5"
)

// MsgPackClient MessagePack RPC 客户端
type MsgPackClient struct {
    baseURL string
    client  *http.Client
}

func NewMsgPackClient(baseURL string) *MsgPackClient {
    return &MsgPackClient{
        baseURL: baseURL,
        client:  &http.Client{Timeout: 5 * time.Second},
    }
}

func (c *MsgPackClient) Call(method string, params interface{}) (map[string]interface{}, error) {
    // 序列化请求
    body, err := msgpack.Marshal(params)
    if err != nil {
        return nil, err
    }

    // 发送请求
    url := fmt.Sprintf("%s/%s", c.baseURL, method)
    req, _ := http.NewRequest("GET", url, bytes.NewReader(body))
    req.Header.Set("Content-Type", "application/x-msgpack")
    req.Header.Set("Accept", "application/x-msgpack")

    resp, err := c.client.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    // 解析响应
    respBody, err := io.ReadAll(resp.Body)
    if err != nil {
        return nil, err
    }

    var result map[string]interface{}
    if err := msgpack.Unmarshal(respBody, &result); err != nil {
        return nil, err
    }

    return result, nil
}

func main() {
    userURL := os.Getenv("USER_SERVICE_URL")
    orderURL := os.Getenv("ORDER_SERVICE_URL")

    userClient := NewMsgPackClient("http://" + userURL)
    orderClient := NewMsgPackClient("http://" + orderURL)

    app := fiber.New()

    // JSON → 内部 MessagePack → JSON
    app.Get("/api/users/:id", func(c *fiber.Ctx) error {
        id := c.Params("id")
        result, err := userClient.Call("users/"+id, nil)
        if err != nil {
            return c.Status(500).JSON(fiber.Map{"error": err.Error()})
        }
        return c.JSON(result)
    })

    // 直接 MessagePack 转发
    app.Get("/api/msgpack/users/:id", func(c *fiber.Ctx) error {
        id := c.Params("id")
        result, err := userClient.Call("users/"+id, nil)
        if err != nil {
            return c.Status(500).Send(msgpackError(err))
        }
        body, _ := msgpack.Marshal(result)
        c.Set("Content-Type", "application/x-msgpack")
        return c.Send(body)
    })

    // 聚合多个服务
    app.Get("/api/dashboard/:userId", func(c *fiber.Ctx) error {
        userId := c.Params("userId")

        // 并行调用
        type userResult struct {
            data map[string]interface{}
            err  error
        }
        type orderResult struct {
            data map[string]interface{}
            err  error
        }

        userCh := make(chan userResult, 1)
        orderCh := make(chan orderResult, 1)

        go func() {
            data, err := userClient.Call("users/"+userId, nil)
            userCh <- userResult{data, err}
        }()

        go func() {
            data, err := orderClient.Call("orders", map[string]interface{}{"userId": userId})
            orderCh <- orderResult{data, err}
        }()

        ur := <-userCh
        or := <-orderCh

        if ur.err != nil || or.err != nil {
            return c.Status(500).JSON(fiber.Map{"error": "服务调用失败"})
        }

        return c.JSON(fiber.Map{
            "user":   ur.data["result"],
            "orders": or.data["result"],
        })
    })

    app.Listen(":8080")
}

func msgpackError(err error) []byte {
    data, _ := msgpack.Marshal(map[string]string{"error": err.Error()})
    return data
}

# shared/cache.py
import msgpack
import redis
import hashlib
import json
from typing import Any, Optional
from functools import wraps

class MsgPackCache:
    """基于 MessagePack 的 Redis 缓存层"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url, decode_responses=False)
        self.default_ttl = 3600  # 1 小时
    
    def _pack(self, value: Any) -> bytes:
        return msgpack.packb(value, use_bin_type=True, default=str)
    
    def _unpack(self, data: bytes) -> Any:
        return msgpack.unpackb(data, raw=False)
    
    def _make_key(self, namespace: str, key: str) -> bytes:
        return f"{namespace}:{key}".encode()
    
    def get(self, namespace: str, key: str) -> Optional[Any]:
        data = self.redis.get(self._make_key(namespace, key))
        if data is None:
            return None
        return self._unpack(data)
    
    def set(self, namespace: str, key: str, value: Any, ttl: int = None):
        packed = self._pack(value)
        self.redis.set(
            self._make_key(namespace, key),
            packed,
            ex=ttl or self.default_ttl
        )
    
    def delete(self, namespace: str, key: str):
        self.redis.delete(self._make_key(namespace, key))
    
    def get_or_set(self, namespace: str, key: str, factory, ttl: int = None) -> Any:
        """获取缓存，不存在则创建"""
        value = self.get(namespace, key)
        if value is not None:
            return value
        value = factory()
        self.set(namespace, key, value, ttl)
        return value

# ========== 缓存装饰器 ==========
def cached(namespace: str, ttl: int = 3600):
    """缓存装饰器"""
    cache = MsgPackCache()
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存键
            key_data = json.dumps({"args": str(args), "kwargs": str(kwargs)}, sort_keys=True)
            cache_key = hashlib.md5(key_data.encode()).hexdigest()
            
            return cache.get_or_set(namespace, cache_key, lambda: func(*args, **kwargs), ttl)
        return wrapper
    return decorator

# 使用示例
@cached("users", ttl=300)
def get_user(user_id: int) -> dict:
    # 模拟数据库查询
    print(f"查询数据库: user_id={user_id}")
    return {"id": user_id, "name": "Alice"}

# 第一次调用: 查询数据库
user = get_user(1)
# 第二次调用: 从缓存获取
user = get_user(1)

import msgpack
import json
import sys

def compare_cache_sizes(data):
    """对比 JSON 和 MessagePack 的缓存大小"""
    json_data = json.dumps(data, ensure_ascii=False).encode("utf-8")
    mp_data = msgpack.packb(data, use_bin_type=True)
    
    print(f"数据: {str(data)[:50]}...")
    print(f"JSON:        {len(json_data):6d} bytes")
    print(f"MessagePack: {len(mp_data):6d} bytes")
    print(f"节省:        {(1 - len(mp_data)/len(json_data))*100:.1f}%")
    print()

# 测试用例
compare_cache_sizes({"id": 1, "name": "Alice", "active": True})
compare_cache_sizes({"users": [{"id": i, "name": f"User{i}"} for i in range(10)]})
compare_cache_sizes({"config": {"debug": True, "timeout": 30, "retries": 3}})

# docker-compose.yml
services:
  gateway:
    build: ./gateway
    volumes:
      - msgpack-socket:/var/run/msgpack
    depends_on:
      - user-service

  user-service:
    build: ./user-service
    volumes:
      - msgpack-socket:/var/run/msgpack

volumes:
  msgpack-socket:

# 使用 mmap 进行同主机高速通信
import mmap
import struct
import msgpack
import os

class SharedMemoryIPC:
    """基于共享内存的 MessagePack IPC"""
    
    def __init__(self, name: str, size: int = 1024 * 1024):
        self.name = f"/msgpack_{name}"
        self.size = size
        
        # 创建共享内存
        self.fd = os.open(f"/dev/shm{self.name}", os.O_CREAT | os.O_RDWR, 0o600)
        os.ftruncate(self.fd, size)
        self.mm = mmap.mmap(self.fd, size)
    
    def write(self, data: dict):
        packed = msgpack.packb(data, use_bin_type=True)
        header = struct.pack(">I", len(packed))
        self.mm.seek(0)
        self.mm.write(header + packed)
    
    def read(self) -> dict:
        self.mm.seek(0)
        header = self.mm.read(4)
        length = struct.unpack(">I", header)[0]
        body = self.mm.read(length)
        return msgpack.unpackb(body, raw=False)
    
    def close(self):
        self.mm.close()
        os.close(self.fd)

# health_check.py
import msgpack
import time
import psutil
from aiohttp import web

async def health_check(request):
    """健康检查端点"""
    health = {
        "status": "healthy",
        "timestamp": int(time.time()),
        "uptime": int(time.time() - start_time),
        "memory": {
            "used_mb": psutil.Process().memory_info().rss / 1024 / 1024,
            "percent": psutil.virtual_memory().percent,
        },
        "connections": active_connections,
    }
    
    # 根据 Accept 头返回格式
    if request.headers.get("Accept") == "application/x-msgpack":
        body = msgpack.packb(health, use_bin_type=True)
        return web.Response(body=body, content_type="application/x-msgpack")
    else:
        return web.json_response(health)

# metrics.py
from prometheus_client import Counter, Histogram, generate_latest
import time

# 定义指标
MSGPACK_REQUESTS = Counter(
    "msgpack_requests_total",
    "MessagePack 请求总数",
    ["method", "status"]
)

MSGPACK_LATENCY = Histogram(
    "msgpack_request_duration_seconds",
    "MessagePack 请求延迟",
    ["method"],
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0]
)

MSGPACK_SIZE = Histogram(
    "msgpack_message_size_bytes",
    "MessagePack 消息大小",
    buckets=[16, 64, 256, 1024, 4096, 16384, 65536]
)

def track_request(method: str):
    """请求追踪装饰器"""
    def decorator(func):
        async def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = await func(*args, **kwargs)
                MSGPACK_REQUESTS.labels(method=method, status="success").inc()
                return result
            except Exception as e:
                MSGPACK_REQUESTS.labels(method=method, status="error").inc()
                raise
            finally:
                MSGPACK_LATENCY.labels(method=method).observe(time.time() - start)
        return wrapper
    return decorator

# ❌ 不好: 使用完整 Python 镜像
FROM python:3.12  # ~900MB

# ✅ 好: 使用 slim 镜像
FROM python:3.12-slim  # ~130MB

# ✅ 更好: 使用 alpine 镜像
FROM python:3.12-alpine  # ~50MB

# docker-compose.yml
services:
  gateway:
    environment:
      - HTTP_TIMEOUT=5  # 5 秒超时
      - RETRY_COUNT=3   # 重试 3 次

# ❌ 不好: 直接打印 MessagePack 二进制
print(f"收到: {packed_data}")  # 乱码

# ✅ 好: 打印十六进制或解码后的内容
print(f"收到: {packed_data.hex()[:50]}...")
# 或
print(f"收到: {msgpack.unpackb(packed_data, raw=False)}")